diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js index 56b819abac4f..847b3837af4c 100644 --- a/lib/internal/streams/readable.js +++ b/lib/internal/streams/readable.js @@ -138,6 +138,7 @@ const kHasPaused = 1 << 25; const kPaused = 1 << 26; const kDataListening = 1 << 27; const kEndScheduled = 1 << 28; +const kEofReadablePending = 1 << 29; // TODO(benjamingr) it is likely slower to do it this way than with free functions function makeBitMapDescriptor(bit) { @@ -682,7 +683,7 @@ Readable.prototype.read = function(n) { state.highWaterMark = computeNewHighWaterMark(n); if (n !== 0) - state[kState] &= ~kEmittedReadable; + state[kState] &= ~(kEmittedReadable | kEofReadablePending); // If we're doing read(0) to trigger a readable event, but we // already have a bunch of data in the buffer, then just trigger @@ -825,7 +826,16 @@ function onEofChunk(stream, state) { // If we are sync, wait until next tick to emit the data. // Otherwise we risk emitting data in the flow() // the readable code triggers during a read() call. - emitReadable(stream); + if ((state[kState] & (kFlowing | kNeedReadable)) !== 0 || + stream.listenerCount('readable') > 0) { + emitReadable(stream); + } else { + // Nobody is observing the stream: do not schedule the 'readable' + // emission at all. A 'readable' listener attached later redeems it + // (see Readable.prototype.on), and read() and resume() reach the + // end of the stream on their own. + state[kState] |= kEofReadablePending; + } } else { // Emit 'readable' now to make sure it gets picked up. state[kState] &= ~kNeedReadable; @@ -1178,8 +1188,17 @@ Readable.prototype.on = function(ev, fn) { debug('on readable'); if (state.length) { emitReadable(this); - } else if ((state[kState] & kReading) === 0) { - process.nextTick(nReadingNextTick, this); + } else { + if ((state[kState] & kEofReadablePending) !== 0) { + // The end-of-stream 'readable' emission was skipped because + // nobody was listening when the stream ended (see onEofChunk): + // emit it now. + state[kState] &= ~kEofReadablePending; + emitReadable(this); + } + if ((state[kState] & kReading) === 0) { + process.nextTick(nReadingNextTick, this); + } } } } diff --git a/test/parallel/test-stream-readable-unobserved-eof-readable.js b/test/parallel/test-stream-readable-unobserved-eof-readable.js new file mode 100644 index 000000000000..5eb6d1b2b631 --- /dev/null +++ b/test/parallel/test-stream-readable-unobserved-eof-readable.js @@ -0,0 +1,76 @@ +'use strict'; +const common = require('../common'); +const { Readable } = require('stream'); +const assert = require('assert'); + +// When a stream reaches its end while nobody is observing it, the +// end-of-stream 'readable' emission is not scheduled. It is emitted +// later if a 'readable' listener is attached, and read()/resume() +// still reach 'end' on their own. + +{ + // Listener attached synchronously after push(null) gets 'readable' + // and then 'end'. + const r = new Readable({ read() {} }); + r.push(null); + let readableEmitted = false; + r.on('readable', common.mustCall(() => { + readableEmitted = true; + assert.strictEqual(r.read(), null); + })); + r.on('end', common.mustCall(() => { + assert.strictEqual(readableEmitted, true); + })); +} + +{ + // Listener attached one macrotask after the unobserved end still gets + // the owed 'readable' before 'end'. + const r = new Readable({ read() {} }); + r.push(null); + setImmediate(common.mustCall(() => { + let readableEmitted = false; + r.on('readable', common.mustCall(() => { + readableEmitted = true; + assert.strictEqual(r.read(), null); + })); + r.on('end', common.mustCall(() => { + assert.strictEqual(readableEmitted, true); + })); + })); +} + +{ + // A stream that ends unobserved still emits 'end' when resumed later. + const r = new Readable({ read() {} }); + r.push(null); + setImmediate(common.mustCall(() => { + r.resume(); + r.on('end', common.mustCall()); + })); +} + +{ + // read() after an unobserved end consumes the owed notification: a + // 'readable' listener attached afterwards does not receive it, but + // 'end' is still emitted. + const r = new Readable({ read() {} }); + r.push(null); + setImmediate(common.mustCall(() => { + assert.strictEqual(r.read(), null); + r.on('end', common.mustCall()); + })); +} + +{ + // A 'data' listener attached after an unobserved end still gets 'end'. + const r = new Readable({ read() {} }); + r.push('x'); + r.push(null); + setImmediate(common.mustCall(() => { + r.on('data', common.mustCall((chunk) => { + assert.strictEqual(chunk.toString(), 'x'); + })); + r.on('end', common.mustCall()); + })); +}