GeezerVison is showing github.com — stream: fix pipeline callback not called on ended stream · nodejs/node@247caac
File treeExpand file treeCollapse file tree Open diff view settings Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: lib/internal/streams/pipeline.js +10-2Lines changed: 10 additions & 2 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -38,6 +38,7 @@ const { | 38 | 38 | isTransformStream, | 39 | 39 | isWebStream, | 40 | 40 | isReadableStream, | | 41 | + isReadableEnded, | 41 | 42 | } = require('internal/streams/utils'); | 42 | 43 | const { AbortController } = require('internal/abort_controller'); | 43 | 44 |
| @@ -417,10 +418,17 @@ function pipe(src, dst, finish, { end }) { | 417 | 418 | // Compat. Before node v10.12.0 stdio used to throw an error so | 418 | 419 | // pipe() did/does not end() stdio destinations. | 419 | 420 | // Now they allow it but "secretly" don't close the underlying fd. | 420 | | - src.once('end', () => { | | 421 | + | | 422 | + function endFn() { | 421 | 423 | ended = true; | 422 | 424 | dst.end(); | 423 | | - }); | | 425 | + } | | 426 | + | | 427 | + if (isReadableEnded(src)) { // End the destination if the source has already ended. | | 428 | + process.nextTick(endFn); | | 429 | + } else { | | 430 | + src.once('end', endFn); | | 431 | + } | 424 | 432 | } else { | 425 | 433 | finish(); | 426 | 434 | } | |
Collapse file Copy file name to clipboardExpand all lines: test/parallel/test-stream-pipeline.js +25Lines changed: 25 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -1591,3 +1591,28 @@ const tsp = require('timers/promises'); | 1591 | 1591 | assert.strictEqual(writable.endCount, 1); | 1592 | 1592 | })); | 1593 | 1593 | } | | 1594 | + | | 1595 | +{ | | 1596 | + const readable = new Readable({ | | 1597 | + read() {} | | 1598 | + }); | | 1599 | + readable.on('end', common.mustCall(() => { | | 1600 | + pipeline(readable, new PassThrough(), common.mustSucceed()); | | 1601 | + })); | | 1602 | + readable.push(null); | | 1603 | + readable.read(); | | 1604 | +} | | 1605 | + | | 1606 | +{ | | 1607 | + const dup = new Duplex({ | | 1608 | + read() {}, | | 1609 | + write(chunk, enc, cb) { | | 1610 | + cb(); | | 1611 | + } | | 1612 | + }); | | 1613 | + dup.on('end', common.mustCall(() => { | | 1614 | + pipeline(dup, new PassThrough(), common.mustSucceed()); | | 1615 | + })); | | 1616 | + dup.push(null); | | 1617 | + dup.read(); | | 1618 | +} |
Collapse file Copy file name to clipboardExpand all lines: test/parallel/test-webstreams-pipeline.js +11-1Lines changed: 11 additions & 1 deletion | Original file line number | Diff line number | Diff line change |
|---|
| 2 | 2 |
| 3 | 3 | const common = require('../common'); | 4 | 4 | const assert = require('assert'); | 5 | | -const { Readable, Writable, Transform, pipeline } = require('stream'); | | 5 | +const { Readable, Writable, Transform, pipeline, PassThrough } = require('stream'); | 6 | 6 | const { pipeline: pipelinePromise } = require('stream/promises'); | 7 | 7 | const { ReadableStream, WritableStream, TransformStream } = require('stream/web'); | 8 | 8 | const http = require('http'); | @@ -410,3 +410,13 @@ const http = require('http'); | 410 | 410 | } | 411 | 411 | c.close(); | 412 | 412 | } | | 413 | + | | 414 | +{ | | 415 | + const rs = new ReadableStream({ | | 416 | + start(controller) { | | 417 | + controller.close(); | | 418 | + } | | 419 | + }); | | 420 | + | | 421 | + pipeline(rs, new PassThrough(), common.mustSucceed()); | | 422 | +} |
|