Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Revert 5541300 and add coverage to pipeline #56278

Closed
wants to merge 2 commits into from
+34 −243
Closed

Revert 5541300 and add coverage to pipeline #56278

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e009e53
Revert "stream: handle generator destruction from Duplex.from()"
jakecastelli Dec 17, 2024
175b3aa
test: add coverage for pipeline
jakecastelli Dec 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 7 additions & 52 deletions lib/internal/streams/duplexify.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,19 +83,15 @@ module.exports = function duplexify(body, name) {
}

if (typeof body === 'function') {
let d;

const { value, write, final, destroy } = fromAsyncGen(body, () => {
destroyer(d);
});
const { value, write, final, destroy } = fromAsyncGen(body);

// Body might be a constructor function instead of an async generator function.
if (isDuplexNodeStream(value)) {
return d = value;
return value;
}

if (isIterable(value)) {
return d = from(Duplexify, value, {
return from(Duplexify, value, {
// TODO (ronag): highWaterMark?
objectMode: true,
write,
Expand All @@ -106,16 +102,12 @@ module.exports = function duplexify(body, name) {

const then = value?.then;
if (typeof then === 'function') {
let finalized = false;
let d;

const promise = FunctionPrototypeCall(
then,
value,
(val) => {
// The function returned without (fully) consuming the generator.
if (!finalized) {
destroyer(d);
}
if (val != null) {
throw new ERR_INVALID_RETURN_VALUE('nully', 'body', val);
}
Expand All @@ -131,7 +123,6 @@ module.exports = function duplexify(body, name) {
readable: false,
write,
final(cb) {
finalized = true;
final(async () => {
try {
await promise;
Expand Down Expand Up @@ -217,12 +208,11 @@ module.exports = function duplexify(body, name) {
body);
};

function fromAsyncGen(fn, destructor) {
function fromAsyncGen(fn) {
let { promise, resolve } = PromiseWithResolvers();
const ac = new AbortController();
const signal = ac.signal;

const asyncGenerator = (async function* () {
const value = fn(async function*() {
while (true) {
const _promise = promise;
promise = null;
Expand All @@ -232,44 +222,9 @@ function fromAsyncGen(fn, destructor) {
if (signal.aborted)
throw new AbortError(undefined, { cause: signal.reason });
({ promise, resolve } = PromiseWithResolvers());
// Next line will "break" the loop if the generator is returned/thrown.
yield chunk;
}
})();

const originalReturn = asyncGenerator.return;
asyncGenerator.return = async function(value) {
try {
return await originalReturn.call(this, value);
} finally {
if (promise) {
const _promise = promise;
promise = null;
const { cb } = await _promise;
process.nextTick(cb);

process.nextTick(destructor);
}
}
};

const originalThrow = asyncGenerator.throw;
asyncGenerator.throw = async function(err) {
try {
return await originalThrow.call(this, err);
} finally {
if (promise) {
const _promise = promise;
promise = null;
const { cb } = await _promise;

// asyncGenerator.throw(undefined) should cause a callback error
process.nextTick(cb, err ?? new AbortError());
}
}
};

const value = fn(asyncGenerator, { signal });
}(), { signal });

return {
value,
Expand Down
191 changes: 0 additions & 191 deletions test/parallel/test-stream-duplex-from.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ const assert = require('assert');
const { Duplex, Readable, Writable, pipeline, PassThrough } = require('stream');
const { ReadableStream, WritableStream } = require('stream/web');
const { Blob } = require('buffer');
const sleep = require('util').promisify(setTimeout);

{
const d = Duplex.from({
Expand Down Expand Up @@ -402,193 +401,3 @@ function makeATestWritableStream(writeFunc) {
assert.strictEqual(d.writable, false);
}));
}

{
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']);

await asyncGenerator.return();
await asyncGenerator.return();
await asyncGenerator.return();
}),
common.mustSucceed(() => {
assert.strictEqual(r.destroyed, true);
})
);
}

{
const r = Readable.from(['foo', 'bar', 'baz']);
pipeline(
r,
Duplex.from(async function(asyncGenerator) {
// eslint-disable-next-line no-unused-vars
for await (const _ of asyncGenerator) break;
}),
common.mustSucceed(() => {
assert.strictEqual(r.destroyed, true);
})
);
}

{
const r = Readable.from(['foo', 'bar', 'baz']);
pipeline(
r,
Duplex.from(async function(asyncGenerator) {
const a = await asyncGenerator.next();
assert.strictEqual(a.done, false);
assert.strictEqual(a.value.toString(), 'foo');
const b = await asyncGenerator.return();
assert.strictEqual(b.done, true);
}),
common.mustSucceed(() => {
assert.strictEqual(r.destroyed, true);
})
);
}

{
const r = Readable.from(['foo', 'bar', 'baz']);
pipeline(
r,
Duplex.from(async function(asyncGenerator) {
// Note: the generator is not even started at this point
await asyncGenerator.return();
}),
common.mustSucceed(() => {
assert.strictEqual(r.destroyed, true);
})
);
}

{
const r = Readable.from(['foo', 'bar', 'baz']);
pipeline(
r,
Duplex.from(async function(asyncGenerator) {
// Same as before, with a delay
await sleep(100);
await asyncGenerator.return();
}),
common.mustSucceed(() => {
assert.strictEqual(r.destroyed, true);
})
);
}

{
const r = Readable.from(['foo', 'bar', 'baz']);
pipeline(
r,
Duplex.from(async function(asyncGenerator) {}),
common.mustCall((err) => {
assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
assert.strictEqual(r.destroyed, true);
})
);
}

{
const r = Readable.from(['foo', 'bar', 'baz']);
pipeline(
r,
Duplex.from(async function(asyncGenerator) {
await sleep(100);
}),
common.mustCall((err) => {
assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
assert.strictEqual(r.destroyed, true);
})
);
}

{
const r = Readable.from(['foo', 'bar', 'baz']);
const d = Duplex.from(async function(asyncGenerator) {
while (!(await asyncGenerator.next()).done) await sleep(100);
});

setTimeout(() => d.destroy(), 150);

pipeline(
r,
d,
common.mustCall((err) => {
assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
assert.strictEqual(r.destroyed, true);
})
);
}

{
const r = Duplex.from(async function* () {
for (const value of ['foo', 'bar', 'baz']) {
await sleep(50);
yield value;
}
});
const d = Duplex.from(async function(asyncGenerator) {
while (!(await asyncGenerator.next()).done);
});

setTimeout(() => r.destroy(), 75);

pipeline(
r,
d,
common.mustCall((err) => {
assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
assert.strictEqual(r.destroyed, true);
assert.strictEqual(d.destroyed, true);
})
);
}

{
const r = Readable.from(['foo']);
pipeline(
r,
Duplex.from(async function(asyncGenerator) {
await asyncGenerator.throw(new Error('my error'));
}),
common.mustCall((err) => {
assert.strictEqual(err.message, 'my error');
assert.strictEqual(r.destroyed, true);
})
);
}

{
const r = Readable.from(['foo', 'bar']);
pipeline(
r,
Duplex.from(async function(asyncGenerator) {
await asyncGenerator.next();
await asyncGenerator.throw(new Error('my error'));
}),
common.mustCall((err) => {
assert.strictEqual(err.message, 'my error');
assert.strictEqual(r.destroyed, true);
})
);
}

{
const r = Readable.from(['foo', 'bar']);
pipeline(
r,
Duplex.from(async function(asyncGenerator) {
await asyncGenerator.next();
await asyncGenerator.throw();
}),
common.mustCall((err) => {
assert.strictEqual(err.code, 'ABORT_ERR');
assert.strictEqual(r.destroyed, true);
})
);
}
27 changes: 27 additions & 0 deletions test/parallel/test-stream-pipeline.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1723,3 +1723,30 @@ tmpdir.refresh();
});
src.destroy(new Error('problem'));
}

{
async function* myAsyncGenerator(ag) {
for await (const data of ag) {
yield data;
}
}

const duplexStream = Duplex.from(myAsyncGenerator);

const r = new Readable({
read() {
this.push('data1\n');
throw new Error('booom');
},
});

const w = new Writable({
write(chunk, encoding, callback) {
callback();
},
});

pipeline(r, duplexStream, w, common.mustCall((err) => {
assert.deepStrictEqual(err, new Error('booom'));
}));
}
Loading