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.