Allow manipulating the generator in Duplex.from()

[matthieusieben]

Fix for #55077

[mcollina]

lgtm with a few nits, good job

[mcollina]

please remove this empty line

[mcollina]

Could you attach this comment to the section it refers to?

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/62727/

[RedYetiDev]

Suggested change

	const r = Readable.from(['foo', 'bar', 'bar']);
	const r = Readable.from(['foo', 'bar', 'baz']);

(for consistency)

[RedYetiDev]

Suggested change

	const r = Readable.from(['foo', 'bar', 'bar']);
	pipeline(
	r,
	Duplex.from(async function(asyncGenerator) {
	const values = await Array.fromAsync(asyncGenerator);
	assert.deepStrictEqual(values, ['foo', 'bar', 'bar']);
	const r = Readable.from(['foo', 'bar', 'baz']);
	pipeline(
	r,
	Duplex.from(async function(asyncGenerator) {
	const values = await Array.fromAsync(asyncGenerator);
	assert.deepStrictEqual(values, ['foo', 'bar', 'baz']);

(for consistency)

[RedYetiDev]

Suggested change

	if (d) destroyer(d);
	destroyer(d);

The destroyer function already checks whether the stream exists:

node/lib/internal/streams/destroy.js

Line 323 in 3c5ceff

if (!stream || isDestroyed(stream)) {

[RedYetiDev]

Suggested change

	destroyer(d, null);
	destroyer(d);

If err is null, it's treated the same as if it's undefined:

node/lib/internal/streams/destroy.js

Line 327 in 3c5ceff

if (!err && !isFinished(stream)) {

[matthieusieben]

Not sure what is wrong with the tests that are not passing:

=== release test-stream-compose ===
Path: parallel/test-stream-compose
/Users/msi/Github/nodejs/node/test/parallel/test-stream-compose.js:401
    assert.strictEqual(err.code, 'ERR_INVALID_RETURN_VALUE');
                           ^

TypeError: Cannot read properties of undefined (reading 'code')
    at Duplex.<anonymous> (/Users/msi/Github/nodejs/node/test/parallel/test-stream-compose.js:401:28)
    at Duplex.<anonymous> (/Users/msi/Github/nodejs/node/test/common/index.js:493:15)
    at Duplex.<anonymous> (node:internal/util:546:20)
    at Duplex.onclose (node:internal/streams/end-of-stream:161:14)
    at Duplex.emit (node:events:507:28)
    at emitCloseNT (node:internal/streams/destroy:148:10)
    at process.processTicksAndRejections (node:internal/process/task_queues:89:21)

Node.js v23.0.0-pre
Command: out/Release/node --expose-internals /Users/msi/Github/nodejs/node/test/parallel/test-stream-compose.js

=== release test-stream-compose-operator ===
Path: parallel/test-stream-compose-operator
node:internal/process/promises:394
    triggerUncaughtException(err, true /* fromPromise */);
    ^

AssertionError [ERR_ASSERTION]: The input did not match the regular expression /boom/. Input:

'AbortError: The operation was aborted'

    at process.processTicksAndRejections (node:internal/process/task_queues:105:5) {
  generatedMessage: true,
  code: 'ERR_ASSERTION',
  actual: AbortError: The operation was aborted
      at destroyer (node:internal/streams/destroy:328:11)
      at node:internal/streams/duplexify:89:7
      at asyncGenerator.return (node:internal/streams/duplexify:245:9)
      at process.processTicksAndRejections (node:internal/process/task_queues:105:5)
      at async /Users/msi/Github/nodejs/node/test/parallel/test-stream-compose-operator.js:63:5
      at async nextAsync (node:internal/streams/from:182:33) {
    code: 'ABORT_ERR'
  },
  expected: /boom/,
  operator: 'rejects'
}

Node.js v23.0.0-pre
Command: out/Release/node /Users/msi/Github/nodejs/node/test/parallel/test-stream-compose-operator.js

[codecov]

Codecov Report

Attention: Patch coverage is 96.36364% with 2 lines in your changes missing coverage. Please review.

Project coverage is 88.24%. Comparing base (3c5ceff) to head (12439c6).
Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
lib/internal/streams/duplexify.js	96.36%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #55096      +/-   ##
==========================================
- Coverage   88.24%   88.24%   -0.01%     
==========================================
  Files         651      651              
  Lines      183877   183925      +48     
  Branches    35858    35866       +8     
==========================================
+ Hits       162266   162307      +41     
+ Misses      14901    14899       -2     
- Partials     6710     6719       +9     

Files with missing lines	Coverage Δ
lib/internal/streams/duplexify.js	96.48% <96.36%> (-0.09%)	⬇️

... and 32 files with indirect coverage changes