TCP, for...on, and Synchronous Subscription

[jhusain]

One of the promises of Observable is that it can allow us to get TCP for push streams via a future theoretical for...on syntax. Here's an example of for...on in action:

async function getStockPriceSpike(name, delta) {
  let symbol = await getSymbol(name);
  let lastPrice, price;
  for(price on getPrices(symbol)) { // getPrices returns never-ending Observable
    if (lastPrice !== undefined && Math.abs(lastPrice - price) > delta) {
      break;
    }
    lastPrice = price;
  }
  return price;
}

Although in the example above the Observable returned by getPrices() would certainly be asynchronous, Observables may be synchronous. If an Observable is synchronous then it is not possible to unsubscribe before receiving data, because the Subscription object is not returned until the subscribe method completes:

// note that if subscribe method delivers values to observer synchronously...
// these values will be sent to the next method before the subscription is returned.
var subscription = observable.subscribe({ next(v) { console.log(v) } });

Given this constraint there will be no way to support the break keyword in a for...on loop. As a consequence we lose TCP. Note the following example:

async function ex() {
  for(let x on syncObservable) {
    if (x > 3) { break; }
    console.log(x); // this would always print the first value
  }
}

In the example above, if syncObservable synchronously notified the numbers 4, 5, and 6 and then completed, we would see the following console output...

4

...instead of this nothing at all.

Unfortunately an inability to synchronously unsubscribe would appear to be a death blow for TCP. If Observables are to be synchronous as we've concluded, we need to support sync unsubscription to enable TCP in the future.

[benlesh]

Unfortunately an inability to synchronously unsubscribe would appear to be a death blow for TCP. If Observables are to be synchronous as we've concluded, we need to support sync unsubscription to enable TCP in the future.

One option is a start handler on observer that gives you the subscription reference. (ala RxJava), as I think we discussed elsewhere on this repo with @benjchristensen and @trxcllnt

Another option is to pass the subscription as a second argument to the next handler on the observer

[zenparsing]

Another option is to pass the subscription as a second argument to the next handler on the observer

That would seem to be a natural fit for this use case, since you are essentially wanting to cancel from within the next handler.

[benlesh]

@zenparsing honestly, thinking about this and talking to @jhusain who was standing over my shoulder two seconds ago, the start handler (the first one I proposed above) is probably a better choice because it allows for synchronous unsubscription before any side-effects occur.

Basically, start is a handler that gets called and passed the subscription at the moment subscribe is called.

myObservable.subscribe({
  start(subscription) {
  },
  next(value) {
  },
});

[benlesh]

... in fact, RxJS 5 was heavily leaning toward implementing the start handler a few months ago, but backburnered it in favor of pushing out more vital features and infrastructure.

[zenparsing]

start handler (the first one I proposed above) is probably a better choice because it allows for synchronous unsubscription before any side-effects occur.

But that's not necessary for the use case presented here.

Besides, that seems kinda weird. In what case would you want to synchronously unsubscribe before any data was delivered? Wouldn't you just not call subscribe in the first place?

An advantage of providing a second parameter to next is that we can use the same strategy for forEach.

[benlesh]

Besides, that seems kinda weird. In what case would you want to synchronously unsubscribe before any data was delivered? Wouldn't you just not call subscribe in the first place?

It's not that much weirder than using Observable to deal with synchronous resources to begin with.

Suppose an Observable is going to animate some element and report the element's current position. If the first step of the animation occurs synchronously, then by the time next is called, it's too late to stop the subscription and prevent the animation. The maxim of Observable is that you shouldn't create side-effects, it's not necessarily that you can't or that users won't. An Observable could just be a set of state changes over things where the state has already changed by the time you're notified. (Think of do for example in RxJS, those are side-effects "upstream" from the subscriber)

[Frikki]

Mixing synchronous and asynchronous is the recipe for disaster.

[RangerMauve]

To be honest, I always thought of observables as something used strictly for Async code. But that might be because of the way I use them in my applications.

[zenparsing]

As of be763e3, added support for an optional start method on the observer. If the subscription is canceled from the start method, the subscriber function is not called.

[zenparsing]

@jhusain Since we now support the start method, can we close this?