Commit 650c9bd
stream: pipeline with end option
Currently pipeline cannot fully replace pipe due
to the missing end option. This PR adds the end
option to the promisified pipeline method.
PR-URL: #40886
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>1 parent 534409d commit 650c9bd
3 files changed
Lines changed: 50 additions & 17 deletions
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -109,7 +109,7 @@ async function* fromReadable(val) { | |||
| 109 | 109 | yield* Readable.prototype[SymbolAsyncIterator].call(val); | |
| 110 | 110 | } | |
| 111 | 111 | ||
| 112 | - async function pump(iterable, writable, finish) { | ||
| 112 | + async function pump(iterable, writable, finish, opts) { | ||
| 113 | 113 | let error; | |
| 114 | 114 | let onresolve = null; | |
| 115 | 115 | ||
@@ -153,7 +153,9 @@ async function pump(iterable, writable, finish) { | |||
| 153 | 153 | } | |
| 154 | 154 | } | |
| 155 | 155 | ||
| 156 | - writable.end(); | ||
| 156 | + if (opts?.end !== false) { | ||
| 157 | + writable.end(); | ||
| 158 | + } | ||
| 157 | 159 | ||
| 158 | 160 | await wait(); | |
| 159 | 161 | ||
@@ -227,17 +229,22 @@ function pipelineImpl(streams, callback, opts) { | |||
| 227 | 229 | const stream = streams[i]; | |
| 228 | 230 | const reading = i < streams.length - 1; | |
| 229 | 231 | const writing = i > 0; | |
| 232 | + const end = reading || opts?.end !== false; | ||
| 230 | 233 | ||
| 231 | 234 | if (isNodeStream(stream)) { | |
| 232 | - finishCount++; | ||
| 233 | - destroys.push(destroyer(stream, reading, writing, (err) => { | ||
| 234 | - if (!err && !reading && isReadableFinished(stream, false)) { | ||
| 235 | - stream.read(0); | ||
| 236 | - destroyer(stream, true, writing, finish); | ||
| 237 | - } else { | ||
| 238 | - finish(err); | ||
| 239 | - } | ||
| 240 | - })); | ||
| 235 | + if (end) { | ||
| 236 | + finishCount++; | ||
| 237 | + destroys.push(destroyer(stream, reading, writing, (err) => { | ||
| 238 | + if (!err && !reading && isReadableFinished(stream, false)) { | ||
| 239 | + stream.read(0); | ||
| 240 | + destroyer(stream, true, writing, finish); | ||
| 241 | + } else { | ||
| 242 | + finish(err); | ||
| 243 | + } | ||
| 244 | + })); | ||
| 245 | + } else { | ||
| 246 | + stream.on('error', finish); | ||
| 247 | + } | ||
| 241 | 248 | } | |
| 242 | 249 | ||
| 243 | 250 | if (i === 0) { | |
@@ -282,14 +289,17 @@ function pipelineImpl(streams, callback, opts) { | |||
| 282 | 289 | then.call(ret, | |
| 283 | 290 | (val) => { | |
| 284 | 291 | value = val; | |
| 285 | - pt.end(val); | ||
| 292 | + pt.write(val); | ||
| 293 | + if (end) { | ||
| 294 | + pt.end(); | ||
| 295 | + } | ||
| 286 | 296 | }, (err) => { | |
| 287 | 297 | pt.destroy(err); | |
| 288 | 298 | }, | |
| 289 | 299 | ); | |
| 290 | 300 | } else if (isIterable(ret, true)) { | |
| 291 | 301 | finishCount++; | |
| 292 | - pump(ret, pt, finish); | ||
| 302 | + pump(ret, pt, finish, { end }); | ||
| 293 | 303 | } else { | |
| 294 | 304 | throw new ERR_INVALID_RETURN_VALUE( | |
| 295 | 305 | 'AsyncIterable or Promise', 'destination', ret); | |
@@ -302,7 +312,7 @@ function pipelineImpl(streams, callback, opts) { | |||
| 302 | 312 | } | |
| 303 | 313 | } else if (isNodeStream(stream)) { | |
| 304 | 314 | if (isReadableNodeStream(ret)) { | |
| 305 | - ret.pipe(stream); | ||
| 315 | + ret.pipe(stream, { end }); | ||
| 306 | 316 | ||
| 307 | 317 | // Compat. Before node v10.12.0 stdio used to throw an error so | |
| 308 | 318 | // pipe() did/does not end() stdio destinations. | |
@@ -314,7 +324,7 @@ function pipelineImpl(streams, callback, opts) { | |||
| 314 | 324 | ret = makeAsyncIterable(ret); | |
| 315 | 325 | ||
| 316 | 326 | finishCount++; | |
| 317 | - pump(ret, stream, finish); | ||
| 327 | + pump(ret, stream, finish, { end }); | ||
| 318 | 328 | } | |
| 319 | 329 | ret = stream; | |
| 320 | 330 | } else { | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -16,11 +16,13 @@ const eos = require('internal/streams/end-of-stream'); | |||
| 16 | 16 | function pipeline(...streams) { | |
| 17 | 17 | return new Promise((resolve, reject) => { | |
| 18 | 18 | let signal; | |
| 19 | + let end; | ||
| 19 | 20 | const lastArg = streams[streams.length - 1]; | |
| 20 | 21 | if (lastArg && typeof lastArg === 'object' && | |
| 21 | 22 | !isNodeStream(lastArg) && !isIterable(lastArg)) { | |
| 22 | 23 | const options = ArrayPrototypePop(streams); | |
| 23 | 24 | signal = options.signal; | |
| 25 | + end = options.end; | ||
| 24 | 26 | } | |
| 25 | 27 | ||
| 26 | 28 | pl(streams, (err, value) => { | |
@@ -29,7 +31,7 @@ function pipeline(...streams) { | |||
| 29 | 31 | } else { | |
| 30 | 32 | resolve(value); | |
| 31 | 33 | } | |
| 32 | - }, { signal }); | ||
| 34 | + }, { signal, end }); | ||
| 33 | 35 | }); | |
| 34 | 36 | } | |
| 35 | 37 | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -1465,5 +1465,26 @@ const tsp = require('timers/promises'); | |||
| 1465 | 1465 | assert.strictEqual(duplex.destroyed, true); | |
| 1466 | 1466 | } | |
| 1467 | 1467 | ||
| 1468 | - run(); | ||
| 1468 | + run().then(common.mustCall()); | ||
| 1469 | + } | ||
| 1470 | + | ||
| 1471 | + { | ||
| 1472 | + const pipelinePromise = promisify(pipeline); | ||
| 1473 | + | ||
| 1474 | + async function run() { | ||
| 1475 | + const read = new Readable({ | ||
| 1476 | + read() {} | ||
| 1477 | + }); | ||
| 1478 | + | ||
| 1479 | + const duplex = new PassThrough(); | ||
| 1480 | + | ||
| 1481 | + read.push(null); | ||
| 1482 | + | ||
| 1483 | + await pipelinePromise(read, duplex, { end: false }); | ||
| 1484 | + | ||
| 1485 | + assert.strictEqual(duplex.destroyed, false); | ||
| 1486 | + assert.strictEqual(duplex.writableEnded, false); | ||
| 1487 | + } | ||
| 1488 | + | ||
| 1489 | + run().then(common.mustCall()); | ||
| 1469 | 1490 | } | |
0 commit comments