diff --git a/lib/stream/promises.js b/lib/stream/promises.js index 986db2e1f8db8a..4f1b16d430dac1 100644 --- a/lib/stream/promises.js +++ b/lib/stream/promises.js @@ -2,21 +2,41 @@ const { Promise, + ArrayPrototypePop, } = primordials; +const { + addAbortSignalNoValidate, +} = require('internal/streams/add-abort-signal'); + +const { + validateAbortSignal, +} = require('internal/validators'); + let pl; let eos; function pipeline(...streams) { if (!pl) pl = require('internal/streams/pipeline'); + let signal; + const lastArg = streams[streams.length - 1]; + if (typeof lastArg === 'object' && 'aborted' in lastArg) { + signal = ArrayPrototypePop(streams); + } return new Promise((resolve, reject) => { - pl(...streams, (err, value) => { + if (signal) { + validateAbortSignal(signal); + } + const pipe = pl(...streams, (err, value) => { if (err) { reject(err); } else { resolve(value); } }); + if (signal) { + addAbortSignalNoValidate(signal, pipe); + } }); } diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js index 78057f9eeffec6..2300f1d5efeeee 100644 --- a/test/parallel/test-stream-pipeline.js +++ b/test/parallel/test-stream-pipeline.js @@ -469,6 +469,83 @@ const net = require('net'); run(); } +{ + // Check aborted signal without values + const pipelinePromise = promisify(pipeline); + async function run() { + const ac = new AbortController(); + const { signal } = ac; + async function* producer() { + ac.abort(); + await Promise.resolve(); + yield '8'; + } + + const w = new Writable({ + write(chunk, encoding, callback) { + callback(); + } + }); + await pipelinePromise(producer, w, signal); + } + + run().catch(common.mustCall((err) => { + assert.strictEqual(err.name, 'AbortError'); + })); +} + +{ + // Check aborted signal after init. + const pipelinePromise = promisify(pipeline); + async function run() { + const ac = new AbortController(); + const { signal } = ac; + async function* producer() { + yield '5'; + await Promise.resolve(); + ac.abort(); + await Promise.resolve(); + yield '8'; + } + + const w = new Writable({ + write(chunk, encoding, callback) { + callback(); + } + }); + await pipelinePromise(producer, w, signal); + } + + run().catch(common.mustCall((err) => { + assert.strictEqual(err.name, 'AbortError'); + })); +} + +{ + // Check pre-aborted signal + const pipelinePromise = promisify(pipeline); + async function run() { + const signal = new EventTarget(); + signal.aborted = true; + async function* producer() { + yield '5'; + await Promise.resolve(); + yield '8'; + } + + const w = new Writable({ + write(chunk, encoding, callback) { + callback(); + } + }); + await pipelinePromise(producer, w, signal); + } + + run().catch(common.mustCall((err) => { + assert.strictEqual(err.name, 'AbortError'); + })); +} + { const read = new Readable({ read() {}