Add support for converting RxJS Observables to Ember Streams

[benlesh]

Improvements in performance might be gained, providing that the following things are true:

1. This PR is merged and Streams are exposed. [FEATURE ember-metal-stream] emberjs/ember.js#9693
2. Streams are confirmed to have a use, such as binding to properties that return streams.

Performance benefits obviously hinge a lot on how Ember binding to streams might work.

[benlesh]

Per a conversation with @mmun, I put this hacked up prototype together of an "RxObservableStream".

Currently, I'm unable to get it working because I think I'm using fooBinding: someStreamObj incorrectly. The binding complains after it tries to use the stream object as a path string. This is true for both Ember canary and 1.9.0:

The basic class looks like:

function RxObservableStream(observable) {
  this.subject = new Rx.BehaviorSubject(observable || Rx.Observable.empty());
  this.source = this.subject.switch();
}

RxObservableStream.prototype = Object.create(Stream.prototype);
RxObservableStream.prototype.constructor = RxObservableStream;

Ember.merge(RxObservableStream.prototype, {
  disposable: undefined,

  source: undefined,

  subject: undefined,

  valueFn: function() {
    if(!this.disposable) {
      this.disposable = this.source.forEach(this._onNext, this._onError, this._onCompleted);
    }

    return this.lastValue;
  },

  _onNext: function(value) {
    this.lastValue = value;
    this.notify();
  },

  _onError: function(err) {
    this.lastValue = null;
    console.error(err);
    this.notify();
  },

  _onCompleted: function() {
    //TODO: is this necessary?
    console.log('complete');
  },


  setSource: function(observable) {
    this.subject.onNext(observable);
  },

  setValue: function(){
    console.log('test');
  },

  _super$destroy: Stream.prototype.destroy,

  destroy: function(){
    if (this._super$destroy()) {
      if(this.disposable) {
        this.disposable.dispose();
      }

      if(this.subject) {
        this.subject.dispose();
      }

      this.disposable = undefined;
      this.source = undefined;
      this.subject = undefined;

      return true;
    }
  }
});

and in use, I was attempting:

App.IndexController = Ember.ObjectController.extend({
  fooBinding: new RxObservableStream(Rx.Observable.interval(1000)),
});

Which (I hoped) would result in {{foo}} updating once per second with "1", "2", "3", ... etc.

No dice. I'm sure I'm doing something wrong here.

[benlesh]

Similarly, I did try:

App.IndexController = Ember.ObjectController.extend({
  fooBinding: 'myThing',

  myThing: new RxObservableStream(Rx.Observable.interval(1000)),
});

But that only resulted in writing out "[object Object]" to the view, which I suppose is to be expected.

Unfortunately, currently my pattern is to chain properties together to create observables, so I'll need to be able to bind to an observable contained by another property:

Ember.IndexController.extend({
   fooBinding: 'mappedThings',

   mappedThings: rxMap('source', function(x) {
     return x + 'mapped',
   }),

   source: function() {
      return Rx.Observable.interval(1000);
   }.property(),
})

[benlesh]

After digging through this for a while. It seems that this might not be a useful thing. AFAICT, simply calling set when I get the next value from an Observable is effectively doing the same thing.

Closing for now.