diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js index fea0e508411c88..7fd5df9f27938c 100644 --- a/lib/internal/streams/duplexify.js +++ b/lib/internal/streams/duplexify.js @@ -209,22 +209,28 @@ function fromAsyncGen(fn) { const signal = ac.signal; const value = fn(async function*() { while (true) { - const { chunk, done, cb } = await promise; + const _promise = promise; + promise = null; + const { chunk, done, cb } = await _promise; process.nextTick(cb); if (done) return; if (signal.aborted) throw new AbortError(); - yield chunk; ({ promise, resolve } = createDeferredPromise()); + yield chunk; } }(), { signal }); return { value, write(chunk, encoding, cb) { - resolve({ chunk, done: false, cb }); + const _resolve = resolve; + resolve = null; + _resolve({ chunk, done: false, cb }); }, final(cb) { - resolve({ done: true, cb }); + const _resolve = resolve; + resolve = null; + _resolve({ done: true, cb }); }, destroy(err, cb) { ac.abort(); diff --git a/test/parallel/test-stream-duplex-from.js b/test/parallel/test-stream-duplex-from.js index 265b61dfd062f9..446768d6eef3e3 100644 --- a/test/parallel/test-stream-duplex-from.js +++ b/test/parallel/test-stream-duplex-from.js @@ -2,7 +2,7 @@ const common = require('../common'); const assert = require('assert'); -const { Duplex, Readable, Writable } = require('stream'); +const { Duplex, Readable, Writable, pipeline } = require('stream'); { const d = Duplex.from({ @@ -118,3 +118,29 @@ const { Duplex, Readable, Writable } = require('stream'); assert.strictEqual(d.readable, false); })); } + +{ + // https://github.com/nodejs/node/issues/40497 + pipeline( + ['abc\ndef\nghi'], + Duplex.from(async function * (source) { + let rest = ''; + for await (const chunk of source) { + const lines = (rest + chunk.toString()).split('\n'); + rest = lines.pop(); + for (const line of lines) { + yield line; + } + } + yield rest; + }), + async function * (source) { + let ret = ''; + for await (const x of source) { + ret += x; + } + assert.strictEqual(ret, 'abcdefghi'); + }, + common.mustCall(() => {}), + ); +}