stream: support passing generator functions into pipeline

[ronag]

Add support for generators and functions in pipeline.

This makes it possible to do the following:

await pipeline(
  async function* () {
    yield await read();
  },
  async function*(source) {
    await for (const chunk of source) {
      yield chunk;
    }
  },
  async function(source) {
    await for (const chunk of source) {
      await write(chunk):
    }
  }
);

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

[mcollina]

I would prefer we did not pass through the stream equivalents for pipeline - I'm not sure if it's feasible. The main reason is performance: streams adds a lot of overhead and a pipeline composed by async iterators can be extremely more performant.

I'm not sure if this is feasible or just a dream.

[ronag]

I'm not sure if this is feasible or just a dream.

Might be feasible. I've updated the PR.

[ronag]

@mcollina: I think you will like this iteration. Now it's possible to do e.g.

let res = '';
pipeline(async function*() {
  await new Promise((resolve) => process.nextTick(resolve));
  yield 'hello';
  yield 'world';
}, async function*(source) {
  for await (const chunk of source) {
    yield chunk.toUpperCase();
  }
}, async function(source) {
  for await (const chunk of source) {
    res += chunk;
  }
}, common.mustCall((err) => {
  assert.strictEqual(err, undefined);
  assert.strictEqual(res, 'HELLOWORLD');
}));

Without passing through the stream equivalents for pipeline.

Still needs some work to ensure edge cases are covered and errors are properly thrown. WIP label please.

[MylesBorins]

@ronag does this still need a backport?

[ronag]

@MylesBorins I've already backported it #31975. Did I miss a label or something?

[MylesBorins]

@ronag thanks! when something has been backported we usually apply a different label. backport-open once it has been opened and backported-to when it has landed

[targos]

Depends at least on #30869 to land on v12.x

[ronag]

@targos: I don't think this should land on v12

[targos]

@ronag why not? I we do not backport the recent stream changes, this subsystem will be really difficult to maintain on v12. There are almost conflicts with every pull request.

[ronag]

@ronag why not? I we do not backport the recent stream changes, this subsystem will be really difficult to maintain on v12. There are almost conflicts with every pull request.

Disregard my previous comment. You are right.