rx support for Angular 2 Async Pipe

rxPipeRegistry.ts

/// <reference path="../../typings/tsd.d.ts" />
///
import {PipeFactory} from 'angular2/src/change_detection/pipes/pipe';
import {async} from 'angular2/src/change_detection/change_detection';
import {NullPipeFactory, Pipe, PipeRegistry, defaultPipes} from 'angular2/change_detection';
import {bind} from 'angular2/di';
import {ObservablePipe} from 'angular2/pipes';
import * as Rx from 'rx';

export function isObservable(obs) {
  return obs && obs instanceof Rx.Observable;
}

export class RxPipe extends ObservablePipe {
  _subscription: any;
  _observable: any;
  constructor(ref) { super(ref); }
  supports(obs) { return isObservable(obs); }
  _subscribe(obs) {
    this._observable = obs;
    this._subscription = obs.subscribe(
      value => this._updateLatestValue(value),
      e => { throw e; }
    );
  }
}

export class RxPipeFactory extends PipeFactory {
  constructor() { super(); }
  supports(obs) { return isObservable(obs); }
  create(cdRef): Pipe { return new RxPipe(cdRef); }
}

export var rxAsync = [ new RxPipeFactory() ].concat(async);


export var rxPipes = Object.assign({}, defaultPipes, {
  'async': rxAsync
});

export var rxPipeRegistry = [
  bind(PipeRegistry).toValue(new PipeRegistry(rxPipes))
];

Patrick! This is awesome and super helpful. There are some changes in 2.0.0.alpha-35 that make this process even easier. I've written up a gist here: https://gist.github.com/jashmenn/d8f5cbf5fc20640bac30

Thankfully the Angular team will be using RxJS natively soon so this workaround wont be needed at all and we can just use the async pipe. However, right now async doesn't detect a Rx.Observable as an Observable so that's why we have to create our own pipe.