stream: add Transform.by utility function

[davidmarkclements]

Analogous to Readable.from, Transform.by creates transform streams
from async function generators.

Checklist

• make -j4 test (UNIX), or vcbuild test (Windows) passes
• tests and/or benchmarks are included
• documentation is changed or added
• commit message follows commit guidelines

[davidmarkclements]

edit: tests passed on a subsequent run (which was a docs fix), guessing its infrastructure failure

the two failing JS tests seem to be failing due to being run against a binary that doesn't have the compiled changes in... maybe it's a caching thing?

[mcollina]

Should we ship this as experimental to start?

[devsnek]

given that https://github.com/tc39/proposal-iterator-helpers is likely to hit stage 2 at the end of this month, and would allow something looking like this:

someReadable[Symbol.asyncIterator]()
	.map(async (x) => { whatever })
	.forEach((item) => writable.push(item));

perhaps instead of adding this api we should just wait for the iterator methods to mature?

[mcollina]

@devsnek I think the two requirements are orthogonal. This PR is all about enabling async iterators and async functions to "play decently" with the Stream ecosystem of modules. In other term, it enables to "use await safely in a transform stream".

The added support for the iterator helpers are needed to improve the language overall, but they won't help in improving the Stream ecosystem.

---

given that https://github.com/tc39/proposal-iterator-helpers is likely to hit stage 2 at the end of this month, and would allow something looking like this:

As a side note, I don't see anything specific to asyncIterator in that repo, and it's not clear to me how we would avoid a file descriptor leak in case the the "map" step blows up.

[devsnek]

@mcollina if that is the case this should probably accept any function. (stream) => stream[Symbol.asyncIterator]().map().filter().etc() should be a valid input.

(also, all the methods being added for sync iterables are also being added for async iterables, and they propogate errors to the end of the chain)

[ronag]

@davidmarkclements: When I try to run this locally the test never finishes:

out/Release/node test/parallel/test-transform-by.js

[ronag]

This probably got broken by f8018f2

[ronag]

I believe this can be implemented in just a few lines using Readble.from, e.g. nxtedition#12. Passes all tests expect for a few ones I'm not sure are needed (e.g. invalid arg type checks should probably be moved into from).

const from = require('internal/streams/from');
Transform.by = function by(asyncGeneratorFn, opts) {
  let _resolve;
  let _promise = new Promise((resolve) => _resolve = resolve);
  return from(Duplex, asyncGeneratorFn(async function*() {
    while (true) {
      const { chunk, done, cb } = await _promise;
      if (done) return cb();
      yield chunk;
      _promise = new Promise((resolve) => _resolve = resolve);
      cb();
    }
  }()), {
    objectMode: true,
    autoDestroy: true,
    ...opts,
    write: (chunk, encoding, cb) => _resolve({ chunk, done: false, cb }),
    final: (cb) => _resolve({ done: true, cb })
  });
};

[ronag]

Any reason why this is not simply called Transform.from similar to Readable.from?

[davidmarkclements]

Any reason why this is not simply called Transform.from similar to Readable.from?

The from to me is more analogous to libraries like from2 - as in we're creating a source stream. So Transform.by is analogous to through2 - it's creating a through stream. We could do Transform.through, but the API symmetry I'm going for is:

• Readable.from
• Transform.by
• Writable.to

[davidmarkclements]

I believe this can be implemented in just a few lines using Readable.from

very cool - I noticed some tests were commented out. I think the one for encoding in particular needs to be supported. Can you speak more to why the others are commented out?

The checks for whether an async function generator is being passed were added at the request of prior reviews, I'd rather honour those initial requests than vacillate on that.

Also, I'd be interested in the performance impact of each approach.

[ronag]

think the one for encoding in particular needs to be supported.

I need to better understand this one. I'll follow up with your comment on the PR.

The checks for whether an async function generator is being passed were added at the request of prior reviews, I'd rather honour those initial requests than vacillate on that.

I agree with them. But I think they should be consistent with and live in Readable.from? Possibly a separate PR? i.e. it's not that we should not have these type checks, just that I think they should live somewhere else.

Also, I'd be interested in the performance impact of each approach.

Maybe a benchmark under /benchmark then?

[ronag]

Also, I'd be interested in the performance impact of each approach.

I made a benchmark but it doesn't work with the version in this PR for some reason? nxtedition@9fbb8d4

[ronag]

I'm not sure what the best course of action here is. @davidmarkclements would you be interested in cherry-picking my proposed changes into this PR?

[ronag]

@mcollina: Is this something we still want considering the recent added functionality in pipeline regarding generators?

[mcollina]

I would say so, IMHO it serves a different purpose.

[ronag]

@davidmarkclements are you still interested in continuing work on this PR or would you mind if I took over?

[jasnell]

Ping @davidmarkclements ... still want to do this?

[davidmarkclements]

@jasnell I've just remembered I was doing this, can we get it in before v14 goes to Active?

[mcollina]

Yes we can!

[safareli]

Hi, this looks super nice!

Is there any plans to get this merged at some point?

[mcollina]

@safareli would you like to work on it? I think it just requires somebody willing to update this and do a fresh PR. I don't think @davidmarkclements will work on this.

[safareli]

@mcollina I've open #38696

[ronag]

Closing in favor of #38696