Debug function for RxJs 6 observable. Using `pipe`

observable.debug.ts

import { MonoTypeOperatorFunction, Observable, Operator, Subscriber, TeardownLogic } from 'rxjs';

/**
 * To debug RxJs 6. Provides 'debug' operator for Observable.
 *
 * example:
 * import { debug } from './debug.observable'
 *
 * myObservable.pipe(
 *    .debug('from my observable')
 * )
 */
class DebugSubscriber<T> extends Subscriber<T> {
    public constructor(private _destination: Subscriber<T>, private message: String) {
        super(_destination);
        console.log(message + ' subscription');
    }

    protected _next(value: T): void {
        console.log(this.message, 'next', value);
        this._destination.next(value);
    }

    protected _error(error: any): void {
        console.log(this.message, 'error', error);
        this._destination.error(error);
    }

    protected _complete(): void {
        console.log(this.message, 'complete');
        this._destination.complete();
    }

    public unsubscribe(): void {
        console.log(this.message, 'unsubscribe');
        super.unsubscribe();
    }
}

export class DebugOperator<T> implements Operator<T, T> {
    public constructor(private message: string) {
        //
    }

    public call(subscriber: Subscriber<T>, source: any): TeardownLogic {
        return source._subscribe(new DebugSubscriber(subscriber, this.message));
    }
}

export const debug = <T>(message: string): MonoTypeOperatorFunction<T> => {
    return (source: Observable<T>) => {
        return source.lift(new DebugOperator<T>(message));
    };
};