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))
];