debug.observable.ts

import {Observable} from "rxjs/Observable"
import {Subscriber} from "rxjs/Subscriber"
import {Operator} from "rxjs/Operator"


/**
 * adds a 'debug' operator to rxjs/Observable
 * 
 * example:
 * import "./debug.observable"
 *
 * myObservable
 *    .debug("from my observable")
 */

function debug<T>(this: Observable<T>, message: string): Observable<T> {
  return this.lift(new DebugOperator<T>(message))
}

class DebugOperator<T> implements Operator<T, T> {

  constructor(private message: string) {
  }

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

class DebugSubscriber<T> extends Subscriber<T> {

  constructor(destination: Subscriber<T>, private message: String) {
    super(destination)
    console.log(message, 'subscription')
  }

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

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

  protected _complete() {
    console.log(this.message, 'complete')
    this.destination.complete()
  }

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

Observable.prototype.debug = debug

declare module 'rxjs/Observable' {
  interface Observable<T> {
    debug: typeof debug
  }
}