RxJS Observable pipe for Angular 2 Alpha 0.40

gistfile1.ts

/**
 * Creates a pipe suitable for a RxJS observable:
 *
 *         @View({
 *           template: '{{ someObservable | rx}}'
 *           pipes: [RxPipe]
 *         })
 *
 * Originally written by @gdi2290 but updated for 2.0.0.alpha-35 and use AsyncPipe
 * (Soon the Angular team will be using RxJS natively and this pipe will be
 * unnecessary because we'll be able to use the `async` pipe.)
 *
 * References:
 * * rxPipeRegistry.ts https://gist.github.com/gdi2290/e9b2880a1d13057197d7 by @gdi2290
 * * AsyncPipe https://github.com/angular/angular/blob/master/modules/angular2/src/pipes/async_pipe.ts
 *
 * @class
 */
import {Pipe, bind, ChangeDetectorRef, Inject, Injectable, NgZone, AsyncPipe} from "angular2/angular2";
import {Observable, Scheduler} from "rx";

function isObservable(obs: any): boolean {
  return obs && typeof obs.subscribe === "function";
}

class RxStrategy {
  constructor(public _zone: NgZone) {}
  createSubscription(async: any, updateLatestValue: any): any {
    return async.observeOn(Scheduler.async).subscribe((val) => { 
      this._zone.run(() => {
        updateLatestValue(val);
      });
    }, (e: Error) => { throw e; });
  }
  dispose(subscription: any): void { subscription.dispose(); }
  onDestroy(subscription: any): void { subscription.dispose(); }
}

@Pipe({name: "rx", pure: false})
export class RxPipe extends AsyncPipe {
  constructor(public _ref: ChangeDetectorRef, public _zone: NgZone) { super(_ref); }

  supports(obs: any): boolean { return isObservable(obs); }

  _selectStrategy(obj: Observable<any>): any {
    return new RxStrategy(this._zone);
  }
}