Implement Symbol.asyncIterator on Observable

[benlesh]

Thinking about the backpressure-related use cases for interop between async iterables and observable, I think I'd consider it an improvement to ergonomics to implement Symbol.asyncIterator on Observable. See this comment here about handling backpressure. There are some really cool/easy/clever things that can be done with this functionality.

Here's a few things to consider:

1. rxjs-for-await has a good amount of usage already.
2. concatMap has exactly the same issue where users need to "understand there is buffering" in some cases, and so far, I haven't seen many people trip over that. In fact, many tutorials and documents steer people towards concatMap for this buffered, one-at-a-time behavior more often than not.
3. for await isn't going away any time soon, and — other than callbacks — is the only real native way to iterate async values (one-at-a-time like concatMap, of course).
4. Subscribing to an observable with for await is obviously non-cancellable, as there's no subscription or even an opportunity to pass a signal or the like, so it's unlikely to "replace" calling subscribe in the hearts and minds of users.
5. Provides even better interop with IxJS and JavaScript in general.

For those new to this, here is what is being proposed (roughly):

for await (const value of someObservable$) {
   await sleep(1000);
   const subValue = await getAnotherValue(value);
   doSomething(subValue);
}

Which would roughly map 1-to-1 with this RxJS behavior:

someObservable$.pipe(
  concatMap(async (value) => {
    await sleep(1000);
    const subValue = await getAnotherValue(value);
    doSomething(subValue);
  })
)
.subscribe();

cc @cartant @kwonoj

[benlesh]

Core Team

Aye: 5, No Opinion: 1 (OJ)