How to convert Node.js stream callback functions into an Async Iterator (jbernier.com)

Just a fun Node.js problem I encountered at work. Wasted time trying to figure this out before realizing the csv-parse library had a Async Iterable API. But thought it was a fun + educational exercise in async generators and async iterables.

Node.js Readable and Web API's ReadableStream both implement [Symbol.asyncIterator], which means you can usually use `for await` without extra code nowadays

  async function* AsAsyncIterable<T>(
    readable: ReadableStream<T>,
  ): AsyncIterable<T> {
    const reader = readable.getReader()
    try {
      while (true) {
        const { done, value } = await reader.read()
        if (done) return
        yield value
      }
    } finally {
      reader.releaseLock()
    }
  }

If it's a readable stream, you can just directly consume it with an async iterator: https://nodejs.org/api/stream.html#streams-compatibility-wit...

I wrote a library a while back which implements this. It handles all the edge cases, provides the ability to measure message backpressure and allows you to close or abruptly kill a stream:

https://github.com/SocketCluster/writable-consumable-stream?...

It has full test coverage and is a core part SocketCluster which has almost full e2e test coverage.

To put it in perspective, the library itself is very lightweight with under 300 lines of code including dependencies (only 1 of them) and has 1.6K lines of test code...

It supports concurrent consumption/iteration by default (each consuming loop causes a new async iterator to be created implicitly in accordance with the AsyncIterator spec) but it also gives you the ability to manually create consumer streams using `stream.createConsumer()` for cases where you want to start buffering messages some time before consuming/iterating over them.