Skip to content

stream: add pipeline() for webstreams #46307

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

Merged
nodejs-github-bot merged 26 commits into from
Feb 2, 2023
Merged

stream: add pipeline() for webstreams #46307

Changes from 1 commit
Commits
Show all changes
26 commits
Select commit Hold shift + click to select a range
4d04a5e
stream: add pipeline() for webstreams
debadree25 Jan 22, 2023
06b8fff
fixup! stop making everything nodestream
debadree25 Jan 25, 2023
08c333a
fixup! final fixups for transform streams
debadree25 Jan 26, 2023
cb7de04
fixup! remove redundant
debadree25 Jan 26, 2023
22ee747
fixup! use proper eos and request type
debadree25 Jan 26, 2023
2ec9a44
fixup! separate out pump function
debadree25 Jan 26, 2023
d1919d9
fixup! generator functions and related tests
debadree25 Jan 26, 2023
6c168d8
fixup! refactor pump functions
debadree25 Jan 26, 2023
68c5c75
fixup! writable stream require no cleanup
debadree25 Jan 26, 2023
4eac7e3
fixup! use pumpWeb
debadree25 Jan 27, 2023
4548998
fixup!
debadree25 Jan 27, 2023
86add31
fixup!
debadree25 Jan 27, 2023
78f0cf3
fixup!
debadree25 Jan 27, 2023
b5879c3
fixup!
debadree25 Jan 27, 2023
6190f6b
fixup!
debadree25 Jan 27, 2023
edfffa6
fixup!
debadree25 Jan 27, 2023
37df364
fixup
debadree25 Jan 27, 2023
120c4fc
fixup! replace with pump to web
debadree25 Jan 27, 2023
cd2a640
fixup! lint
debadree25 Jan 27, 2023
7d8414f
fixup! possible backpressure fix
debadree25 Jan 27, 2023
e3c139a
fixup! possible test for backpressure
debadree25 Jan 28, 2023
c1e0241
fixup! update highWatermark test
debadree25 Jan 28, 2023
e37052b
Update lib/internal/streams/pipeline.js
debadree25 Jan 28, 2023
d2fdac4
fixup
debadree25 Jan 29, 2023
4ac3dc3
fixup! add doc
debadree25 Jan 29, 2023
4c2c2eb
fixup! nit on test
debadree25 Jan 30, 2023
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
Prev Previous commit
Next Next commit
fixup! separate out pump function
  • Loading branch information
@debadree25
debadree25 committed Jan 26, 2023
commit 2ec9a448a565900799368aeb3706960f26e01a91
74 changes: 69 additions & 5 deletions lib/internal/streams/pipeline.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@ const {
isWritableStream,
isTransformStream,
isWebStream,
isReadableStream,
} = require('internal/streams/utils');
const { AbortController } = require('internal/abort_controller');

Expand Down Expand Up @@ -98,6 +99,65 @@ async function pump(iterable, writable, finish, { end }) {
let error;
let onresolve = null;

const resume = (err) => {
if (err) {
error = err;
}

if (onresolve) {
const callback = onresolve;
onresolve = null;
callback();
}
};

const wait = () => new Promise((resolve, reject) => {
if (error) {
reject(error);
} else {
onresolve = () => {
if (error) {
reject(error);
} else {
resolve();
}
};
}
});

writable.on('drain', resume);
const cleanup = eos(writable, { readable: false }, resume);

try {
if (writable.writableNeedDrain) {
await wait();
}

for await (const chunk of iterable) {
if (!writable.write(chunk)) {
await wait();
}
}

if (end) {
writable.end();
}

await wait();

finish();
} catch (err) {
finish(error !== err ? aggregateTwoErrors(error, err) : err);
} finally {
cleanup();
writable.off('drain', resume);
}
}

async function pumpWeb(iterable, writable, finish, { end }) {
let error;
let onresolve = null;

if (isTransformStream(writable)) {
writable = writable.writable;
}
Expand Down Expand Up @@ -353,12 +413,13 @@ function pipelineImpl(streams, callback, opts) {
if (isReadable(stream) && isLastStream) {
lastStreamCleanup.push(cleanup);
}
} else if (isTransformStream(ret) || isReadableStream(ret)) {
const toRead = ret.readable || ret;
finishCount++;
pumpWeb(toRead, stream, finish, { end });
} else if (isIterable(ret)) {
finishCount++;
pump(ret, stream, finish, { end });
} else if (isTransformStream(ret)) {
finishCount++;
pump(ret.readable, stream, finish, { end });
} else {
throw new ERR_INVALID_ARG_TYPE(
'val', ['Readable', 'Iterable', 'AsyncIterable'], ret);
Expand All @@ -368,11 +429,14 @@ function pipelineImpl(streams, callback, opts) {
if (isReadableNodeStream(ret)) {
finishCount += 2;
pipeNodeToWeb(ret, stream, finish, { end });
} else if (isReadableStream(ret)) {
finishCount++;
pumpWeb(ret, stream, finish, { end });
} else if (isTransformStream(ret)) {
pumpWeb(ret.readable, stream, finish, { end });
} else if (isIterable(ret)) {
finishCount++;
pump(ret, stream, finish, { end });
} else if (isTransformStream(ret)) {
pump(ret.readable, stream, finish, { end });
} else {
throw new ERR_INVALID_ARG_TYPE(
'val', ['Readable', 'Iterable', 'AsyncIterable'], ret);
Expand Down