Node: async iterators/generators in stream.pipeline()

[boneskull]

Apologies for all the code here.

We can consume a Readable stream using an async iterator:

// source: http://2ality.com/2018/04/async-iter-nodejs.html
async function printAsyncIterable(iterable) {
  for await (const chunk of iterable) {
    console.log('>>> '+chunk);
  }
}

printAsyncIterable(fs.createReadStream('my-file.txt', 'utf8'));

And we can use async generators similarly to how one would use a Transform stream:

/**
 * Parameter: async iterable of chunks (strings)
 * Result: async iterable of lines (incl. newlines)
 */
async function* chunksToLines(chunksAsync) {
  let previous = '';
  for await (const chunk of chunksAsync) {
    previous += chunk;
    let eolIndex;
    while ((eolIndex = previous.indexOf('\n')) >= 0) {
      // line includes the EOL
      const line = previous.slice(0, eolIndex+1);
      yield line;
      previous = previous.slice(eolIndex+1);
    }
  }
  if (previous.length > 0) {
    yield previous;
  }
}

/**
 * Parameter: async iterable of lines
 * Result: async iterable of numbered lines
 */
async function* numberLines(linesAsync) {
  let counter = 1;
  for await (const line of linesAsync) {
    yield counter + ': ' + line;
    counter++;
  }
}

Then, we can "pipe" these together like so:

async function main() {
  printAsyncIterable(
    numberLines(
      chunksToLines(
        fs.createReadStream('my-file.txt', 'utf8')
      )
    )
  );
}
main();

That's neat, but also kind of hideous. What if we could leverage stream.pipeline() to do something like this?

async function main() {
  stream.pipeline(
    fs.createReadStream('my-file.txt', 'utf8'),
    chunksToLines,
    numberLines,
    printAsyncIterable
  );
}
main();

I'm unfamiliar with the guts of stream.pipeline()--and completely new to async iterators and generators--so don't know how feasible something like this is.

FWIW, the "hideous nested function calls" can be naively replaced by use of the godlike Array.prototype.reduce():

const pipeline = async (...args) => args.reduce((acc, arg) => arg(acc));

async function main() {
  pipeline(
    fs.createReadStream('my-file.txt', 'utf8'),
    chunksToLines,
    numberLines,
    printAsyncIterable
  );
}
main();

Reference: https://twitter.com/b0neskull/status/1115325542566227968

[boneskull]

cc @mcollina

[addaleax]

So … basically, the desired feature here is that stream.pipeline() should transparently convert async generators to object-mode Transform streams?

[boneskull]

@addaleax I guess that'd be pretty great. Readable#pipe too, maybe?

[boneskull]

something something migration path from streams to async iterators and generators

[devsnek]

if we can coerce async iterables to streams, that would enable some nice compat with whatwg streams. +1 from me

[mcollina]

How can we coerce to a transform? What signature would you expect from a transform async iterator?

[addaleax]

@mcollina I think any function that takes one argument, which is an async-iterable for Buffers or strings, and returns an async-iterable of that kind should work?

More generally, I think we could create an utility class that turns async-iterables of Buffers/strings into a Readable stream, and use that as a building block for the .pipe() utility.

[mcollina]

https://github.com/mcollina/stream-iterators-utils/blob/master/README.md#utilstoreadablegenerator-opts this is a prototype implementation of that Readable pattern.

[felixfbecker]

It's awesome that we now have streams.Readable.from() and readable[Symbol.asyncIterator], the only piece missing in the puzzle now is streams.Transform.from() with a signature like

class Transform<I, O> {
  static from<I, O>(generatorFn: (input: AsyncIterable<I>) => AsyncIterable<O>): Transform<I, O>
}

[ronag]

We added support for async generators in pipeline in v13.x. I believe that full-fills this topic. Please re-open of there is anything further to address.