Skip to content

Commit

Permalink
stream: fix fromAsyncGen
Browse files Browse the repository at this point in the history
  • Loading branch information
ronag committed Oct 18, 2021
1 parent c0a7020 commit 226ff00
Show file tree
Hide file tree
Showing 2 changed files with 31 additions and 2 deletions.
4 changes: 3 additions & 1 deletion lib/internal/streams/duplexify.js
Original file line number Diff line number Diff line change
Expand Up @@ -210,11 +210,13 @@ function fromAsyncGen(fn) {
const value = fn(async function*() {
while (true) {
const { chunk, done, cb } = await promise;
promise = null;
resolve = null;
process.nextTick(cb);
if (done) return;
if (signal.aborted) throw new AbortError();
yield chunk;
({ promise, resolve } = createDeferredPromise());
yield chunk;
}
}(), { signal });

Expand Down
29 changes: 28 additions & 1 deletion test/parallel/test-stream-duplex-from.js
Original file line number Diff line number Diff line change
Expand Up @@ -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({
Expand Down Expand Up @@ -118,3 +118,30 @@ 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) {
console.error(0)
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(() => {}),
)
}

0 comments on commit 226ff00

Please sign in to comment.