stream: skip unobserved 'readable' emission at EOF

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);
+        }
       }
     }
   }

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());
+  }));
+}