Observables can be made async iterable by using a buffer

[bakkot]

RxJS observables are async iterable (implemented using an unbounded buffer).

There's some good discussion of considerations and tradeoffs linked from that thread. Something to consider here.

// click-and-drag handler
element.addEventListener('mousedown', async () => {
  for await (let { clientX, clientY } of element.on('mousemove').takeUntil(element.on('mouseup')) {
    console.log('moved mouse to', clientX, clientY);
    if (clientX > threshold) {
      break; // triggers `return` handler from the async iterator, which automatically unsubscribes
    }
  }
  console.log('mouse up or exceeded threshold');
});

is pretty cute.

[Jamesernator]

I have found this useful in the past, however I do think it would probably be worth forcing users to pick a queueing strategy rather than just being async iterable by default. Like depending on what's done with the events I've found all of these queueing strategies useful:

• Single-buffered, i.e. only the latest event is buffered between calls to .next()
  • This is more appropriate when only the most recent event/value is relevant, e.g. things like progress monitoring or following mouse position generally only need most recent event
• Fully buffered
  • This is more appropriate when all intermediate values are important, e.g. for drawing a line with the mouse/pointer then all intermediate events are important to be part of the line
• Unbuffered
  • This is appropriate where values should be ignored until the previous one is processed, e.g. in a game if some player action is performed it may simply ignore inputs until that action has finished resolving

[benlesh]

FWIW, flatMap (which is concatMap in RxJS terms) uses the exact same strategy as conversion to an async iterable does. They really aren't any different.

source.flatMap(async (v) => {
  await getSomething(v);
})
.subscribe({
  next: sideEffect
})

// is functionally identical to

for await (const v of source) {
  const result = await getSomething(v);
  sideEffect(result);
}

The major differences being that you're not allocating a Promise<IteratorResult<T>> per turn, and cancellation in the async iterable's case relies on being able to hit a break; line in the for await loop.

[benlesh]

FWIW: I'm 100% in favor of interop between different types. It's very useful.

[benlesh]

@domfarolino I honestly think this is the number one future enhancement we can provide. It's great because it's non-breaking, and there's already a decent amount of prior art for this. In particular, rxjs-for-await had enough usage that RxJS wanted to implement Symbol.asyncIterator on observable.

The default strategy was determined to be "each value from". That is to say you get one value per turn. This was because:

1. Array.fromAsync(observable) should end up being an array of each emitted value. No other strategy would do this cleanly.
2. AsyncIterator.from(observable) should give you an async iterator of each emitted value. No other strategy would do this cleanly.
3. Other strategies can be provided in more explicit ways by userland libraries like RxJS.
4. Most usage of rxjs-for-await indicated that the function eachValueFrom was the most used strategy.

Implementation details that matter: The returned AsyncIterator really needs to be an AsyncGenerator, such that calling return() or throw() aborts the underlying observable subscription. This will ensure that for await loops that break or thrown errors will abort the observable subscription under the hood.