Implement [Symbol.observable] for ES7 interop compliance

[benlesh]

I've already reached out to the Most.js community so a lot of this will be repeated information.... Please excuse me if some of this has been copied and pasted. ;)

I know that a few months ago we discussed trying to come up with an interop spec, and I think we may have something with what has been developed as part of the ES7 Observable spec.

We've been working hard on the next version of RxJS, and we're getting closer. Part of that work was to try to adhere to the current ES7 Observable spec by @zenparsing. The part of particular interest there for this library is the [Symbol.observable]() method.

Impl ideas?

Sadly, I'm afraid I don't know enough about Bacon's API to help you come up with a solid implementation. But I'm reaching out to you because I think Bacon has obvious merit and a strong community. What I really want is more support for streaming types throughout the land.

Consuming "lowercase-o" observables...

On your end, you could implement something inside of your from method to handle any type with a [Symbol.observable] method on it. As a guineapig you can use @reactivex/rxjs over npm, if you choose. Again it's pretty basic. You'd just call [Symbol.observable]() then call subscribe(observer) on the returned object, where observer has next, error and complete methods.

Other things...

I've actively tried to make any monadic-bind-esque operator in RxJS Next support objects that implement Symbol.observable. This means, effectively that if you were to implement this, people could jump between Bacon and RxJS using flatMaps and switches, etc.

[raimohanska]

I don't see any problem in adding some support into Bacon.js, including

• Bacon.Observable.prototype.[Symbol.observable]() for making Bacon observables consumable by other Observable implementation
• Bacon.fromObservable for wrapping any Observable into a Bacon EventStream.
• Support ES7 Observables in Bacon.Observable.prototype.flatMap et al as a fallback in the case the returned object is not a Bacon Observable

These additions would in fact benefit the cases where you have different versions of Bacon running in the same environment too. There's a related issue...

[benlesh]

Awesome! Let me know if you need anything.

[raimohanska]

I need somebody to actually do this :)

Cannot assign a very high priority to this, as I don't see ES7 Observable as something that will be useful very soon. I'm willing to discuss this though, at least. A couple of issues came to my mind:

1) Not entirely sure how the ES7 Observable and Bacon Observable semantics match. For instance, is an Observable supposed to terminate on first error event? In Bacon.js this is not the case; after and Error, there may be more Error or Next events until an End event is emitted. I got the impression that the spec is based on Rx.

2) The spec seems very procedural and lists a lot of assertions etc. Wouldn't like to implement all that. Maybe we could use a reference impl / shim for wrapping Bacon observables into ES7 Observables?

[benlesh]

For instance, is an Observable supposed to terminate on first error event?

Yes. However, you can implement your adapter however you want. Perhaps you collect all of your errors and error once you complete? Or perhaps you just have it error and stop on the first one in that case? I'd say probably the later, since you're going from Bacon -> Observable in that case.

The spec seems very procedural and lists a lot of assertions etc. Wouldn't like to implement all that. Maybe we could use a reference impl / shim for wrapping Bacon observables into ES7 Observables?

There is a polyfill implementation as part of that spec repository you can use. Otherwise, you could even do something very simple and assume the observer being passed in has the proper assertions in place (because it really should).

.. there's actually two sides to this of course. You can consume/convert observables to Bacon observables as well. For that you'll need to implement a basic observer with some guarantees for things like making sure you don't next after an error (because observables shouldn't do that)

class Observer {
  constructor() {
    this.isUnsubscribed = false;
  }

  next(value) {
    if(!this.isUnsubscribed) {
       // do what your going to do here
    }
  }

  error(err) {
     if(!this.isUnsubscribed) {
        this.isUnsubscribed = true;
        // handle the error
     }
   }

   complete() {
      if(!this.isUnsubscribed) {
         this.isUnsubscribed = true;
         // handle completion
      }
   }
}

I got the impression that the spec is based on Rx.

Yes and no. Rx had a heavy influence no doubt, but the spec itself originally started as a "dual" of Iterable. So instead of an iterator method that gave you an iterator, there was an observer method that you provided an observer. The observer had three callbacks rather than three methods, etc. FWIW, it's my understanding this is how Rx.NET evolved as well, which is the granddaddy to RxJS.

I don't see ES7 Observable as something that will be useful very soon.

The Observable itself, probably not, I agree... but the interop points are valuable I think. I know that Angular will be using RxJS internally, but that some of their bindings could (and should) use Symbol.observable. I've also been a part of a few conversations discussing using the interop points from the spec to allow bindings in Ember. Interop also provides some interesting opportunities for Bacon, Rx, and Most to coexist. For example, if Bacon observables had Symbol.observable implemented, a user could flatMap over an RxJS.Observable of Bacon.Observable the way that I'm implementing such operators in RxJS Next.

[benlesh]

I need somebody to actually do this :)

I don't suspect it will take long. I just don't know my way around your library. I'm happy to pair sometime, perhaps?

In the end, I think having interop and some common standard between streaming types will only serve to boost interest in all libraries. It's not really a contest. It's more of a matter of getting people to stop writing crazy imperative code and start thinking reactively.

[zenparsing]

@raimohanska

I don't think @blesh is suggesting the Bacon implement the complete ES7 spec. The only requirement for interop is that Bacon observables:

• Must have a Symbol.observable method

The returned object must have:

• A subscribe method which accepts an observer.

The returned object should have:

• A Symbol.observable method (which will usually return itself).

With that in place, any other "observable" implementation (including ES7) can then convert from Bacon observables:

let es7Observable = Observable.from(baconObservable);

And it also gives Bacon a hook for converting from other observable-ish things.

I'll try to take a look at the codebase to see if I can come up some concrete suggestions.

[zenparsing]

If I'm reading the code correctly, it looks like Bacon's subscribe takes a single callback and returns a cancellation function. This is pretty close to the needed subscribe signature, except for the argument type (which needs to be an object).

You could write an adapter along these lines:

class InteropAdapter {
    constructor(observable) { this._observable = observable }
    subscribe(sink) {
        return this._observable.subscribe(event => {
            if (event.isError()) sink.error(event.error);
            else if (event.isEnd()) sink.complete();
            else sink.next(event);
        });
    }
    [Symbol.observable]() { return this }
}

Bacon.Observable.prototype[Symbol.observable] = function() {
    return new InteropAdapter(this);
};

Perhaps you could also overload Bacon's subscribe function to also take an observer object, and then just do:

Bacon.Observable.prototype[Symbol.observable] = function() {
    return this;
};

But that would be more invasive.

[raimohanska]

@zenparsing thanks, your InteropAdapter looks good! Except we should end on the first Error event from the Bacon.js Observable (Bacon.js Observables don't end on first error).

I think it might actually be worth considering to extend Bacon subscribe to support ES7 Observers too. There's no conflict, because currently it expects the argument to be a function. If that check fails, it can treat the argument as ES7 Observer.

[raimohanska]

In the end, I think having interop and some common standard between streaming types will only serve to boost interest in all libraries. It's not really a contest. It's more of a matter of getting people to stop writing crazy imperative code and start thinking reactively.

Agreed! I think this is a good idea and I don't see any major obstacles for implementing this right away.

Anyone want to take the first shot? I'm keen to see how much polyfilling we need to make it work on current ES5-level environments. I guess Observable and Symbol would be the minimum.

[raimohanska]

So far Bacon.js has been developed so that it should work also in pretty old browsers without polyfills.

[benlesh]

You should be able to get by with a partial observable, which is just an object with a subscribe method and the symbol on it that returns itself. Think like a minimal iterator impl.

[lautis]

I might have some time to work on this.

To avoid dependencies to any polyfills, we could have the Symbol.observable method to be defined only when Symbol.observable exists. Looking at the InteropAdapter and the ES7 observable spec, it's not immediately obvious to me that an Observable polyfill would be needed either (in Bacon). But I might be a bit naïve.

[zenparsing]

it's not immediately obvious to me that an Observable polyfill would be needed either (in Bacon)

@lautis That's right.

Only adding the the method if Symbol.observable is defined globally would be the safe, easy way to go.

[rpominov]

Should the observable returned from stream[Symbol.observable]() be zalgo-safe? And should it catch exceptions thrown from observer.next(), observer.complete(), observer.error()? If yes, what should we do on an exception from each of those methods?

[lautis]

Should the observable returned from stream[Symbol.observable]() be zalgo-safe?

My interpretation of the spec is that emitting events during the same tick is allowed. Only Observable.of(args...) and Observable.from(iterable) are specified to return zalgo-safe streams.

And should it catch exceptions thrown from observer.next(), observer.complete(), observer.error()? If yes, what should we do on an exception from each of those methods?

The reference implementation catches errors, cleans up the stream and re-throws the exception. My assumption is that this behaviour should be matched in the Bacon subscription as well if the stream should adhere to ES7 observable semantics. But the ES7 observable implementations are already supposed to do this, so I did overlook this in the initial stab for getting this implemented.

[rpominov]

But the ES7 observable implementations are already supposed to do this, so I did overlook this in the initial stab for getting this implemented.

You mean there is no issue if we do EsObservable.from(baconStream)? But what if we do, for instance, Kefir.fromObservable(baconStream)?

[lautis]

You mean there is no issue if we do EsObservable.from(baconStream)?

I'm sure there are issues, but in a simple case the EsObservable behaves as described and Bacon stream subscription is cleaned up. This seems pretty reasonable, although I didn't notice any error or end events being emitted.

But what if we do, for instance, Kefir.fromObservable(baconStream)?

With the proposed implementation, exception handling would be left as the responsibility of Kefir. There could be problems if the Bacon stream catches exceptions and stops the stream subscription, but Kefir stream still waits for events.