Translate brook promise and channel into RxJS.

brook-rxjs-bridge.js

Namespace('brook.rxjsbridge')
.use('brook promise')
.define(function(ns) {
  'use strict';
  var observableFromPromise = function(promise, value) {
    return Rx.Observable.create(function(observer) {
      var called = false;
      var throwIfCalled = function() {
        if (called) {
          throw new Error('promise resolved twice.');
        }
        called = true;
      };

      var observerPromise = ns.promise(function(n, v) {
        throwIfCalled();
        if (observer) {
          observer.onNext(v);
          observer.onComplete();
        }
        n(v);
      });
      observerPromise.setErrorHandler(function(e) {
        throwIfCalled();
        if (observer) {
          observer.onError(e);
        }
      });
      observerPromise.run(value);
      return function() {
        observer = null;
      };
    });
  };

  var observableFromChannel = function(channel) {
    return Rx.Observable.create(function(observer) {
      var observerPromise = ns.promise(function(n, v) {
        observer.onNext(v);
        n(v);
      });
      channel.observe(observerPromise);
      return function() {
        channel.stopObserving(observerPromise);
      };
    });
  };

  ns.provide({
    observableFromPromise: observableFromPromise,
    observableFromChannel: observableFromChannel
  });
});