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

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.

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

Might be feasible. I've updated the PR.

@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.

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I’m lost on why this is semver-major. Can you recap why on the PR description?

I think it was in an earlier version but I can't see either that it would be now.

@codebytere: Yes, it seems eos is calling the callback earlier in 13.x. I'll make a fix for master which we can backport.

@ronag does this still need a backport?

@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

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

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

@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 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.