lib: ensure readable stream flows to end

lib/_stream_readable.js

@@ -568,16 +568,38 @@ function maybeReadMore(stream, state) {
 }
 
 function maybeReadMore_(stream, state) {
-  var len = state.length;
+  // Attempt to read more data if we should.
+  //
+  // The conditions for reading more data are (one of):
+  // - Not enough data buffered (state.length < state.highWaterMark). The loop
+  //   is responsible for filling the buffer with enough data if such data
+  //   is available. If highWaterMark is 0 and we are not in the flowing mode
+  //   we should _not_ attempt to buffer any extra data. We'll get more data
+  //   when the stream consumer calls read() instead.
+  // - No data in the buffer, and the stream is in flowing mode. In this mode
+  //   the loop below is responsible for ensuring read() is called. Failing to
+  //   call read here would abort the flow and there's no other mechanism for
+  //   continuing the flow if the stream consumer has just subscribed to the
+  //   'data' event.
+  //
+  // In addition to the above conditions to keep reading data, the following
+  // conditions prevent the data from being read:
+  // - The stream has ended (state.ended).
+  // - There is already a pending 'read' operation (state.reading). This is a
+  //   case where the the stream has called the implementation defined _read()
+  //   method, but they are processing the call asynchronously and have _not_
+  //   called push() with new data. In this case we skip performing more
+  //   read()s. The execution ends in this method again after the _read() ends
+  //   up calling push() with more data.
   while (!state.reading && !state.ended &&
-         state.length < state.highWaterMark) {
+         (state.length < state.highWaterMark ||
+          (state.flowing && state.length === 0))) {
+    const len = state.length;
     debug('maybeReadMore read 0');
     stream.read(0);
     if (len === state.length)
       // didn't get any data, stop spinning.
       break;
-    else
-      len = state.length;
   }
   state.readingMore = false;
 }

test/parallel/test-stream-readable-flow-no-flow-mixed.js

@@ -0,0 +1,102 @@
+'use strict';
+
+const common = require('../common');
+
+// This test ensures that Readable stream switches between flowing and
+// non-flowing state properly when varying the 'readable' and 'data' event
+// subscription.
+
+const assert = require('assert');
+const { Readable } = require('stream');
+
+const flowingData = [
+  { value: 'a' },
+  { value: 'b' },
+  { value: 'c', subscribeData: true },
+  { value: 'd' },
+  { value: 'e' },
+  { value: 'f', removeReadable: true },
+  { value: 'g' },
+  { value: 'h' },
+  { value: 'i', subscribeReadable: true },
+  null,
+];
+
+const r = new Readable({
+  read: common.mustCall(() => {
+    process.nextTick(() => {
+      r.push(flowingData.shift());
+    });
+  }, flowingData.length),
+  objectMode: true,
+
+  // The water mark shouldn't matter but we'll want to ensure the stream won't
+  // buffer data before we have a chance to react to the subscribe/unsubscribe
+  // event controls.
+  highWaterMark: 0,
+});
+
+// Store data received through 'readable' events and 'data' events.
+const actualReadable = [];
+const actualData = [];
+
+r.on('end', common.mustCall(() => {
+  assert.deepStrictEqual(actualReadable, ['a', 'b', 'c', 'd', 'e', 'f']);
+  assert.deepStrictEqual(actualData, ['d', 'e', 'f', 'g', 'h', 'i']);
+}));
+
+// Subscribing 'readable' should set flowing state to false.
+assert.strictEqual(r.readableFlowing, null);
+r.on('readable', common.mustCall(() => {
+  const v = r.read();
+  actualReadable.push(v.value);
+
+  if (v.subscribeData) {
+
+    // Subsribing 'data' should not change flowing state.
+    assert.strictEqual(r.readableFlowing, false);
+    r.on('data', common.mustCall((data) => {
+      actualData.push(data.value);
+
+      if (data.subscribeReadable) {
+
+        // Re-subsribing readable should put the stream back to non-flowing
+        // state.
+        assert.strictEqual(r.readableFlowing, true);
+        r.on('readable', common.mustCall(() => {
+          // The stream is at the end, but 'readable' is signaled without the
+          // stream knowing this. The 'r.read()' here will result in _read
+          // getting executed, which will then push the final null.
+          //
+          // NOTE: The 'null' here signals non-synchronous read. It is NOT the
+          // same 'null' that the _read ends up pushing to signal end of
+          // stream.
+          assert.strictEqual(r.read(), null);
+        }));
+        assert.strictEqual(r.readableFlowing, false);
+      }
+    }, 6));
+    assert.strictEqual(r.readableFlowing, false);
+  }
+
+  if (v.removeReadable) {
+    // Removing 'readable' should allow the stream to flow into 'data' without
+    // us calling 'read()' manually.
+    //
+    // This should also cahgne the flowing state - although it is delayed into
+    // the next tick (within removeAllListeners).
+    assert.strictEqual(r.readableFlowing, false);
+    r.removeAllListeners('readable');
+    process.nextTick(() => {
+      assert.strictEqual(r.readableFlowing, true);
+    });
+  } else {
+    // We'll need to call r.read() to trigger the next read.
+    //
+    // It should return 'null' as the actual _read implementation is
+    // asynchronous but we still need to call it to trigger the push on
+    // next tick.
+    assert.strictEqual(r.read(), null);
+  }
+}, 6));
+assert.strictEqual(r.readableFlowing, false);

test/parallel/test-stream-readable-hwm-0-async.js

@@ -0,0 +1,27 @@
+'use strict';
+
+const common = require('../common');
+
+// This test ensures that Readable stream will continue to call _read
+// for streams with highWaterMark === 0 once the stream returns data
+// by calling push() asynchronously.
+
+const { Readable } = require('stream');
+
+let count = 5;
+
+const r = new Readable({
+  // Called 6 times: First 5 return data, last one signals end of stream.
+  read: common.mustCall(() => {
+    process.nextTick(common.mustCall(() => {
+      if (count--)
+        r.push('a');
+      else
+        r.push(null);
+    }));
+  }, 6),
+  highWaterMark: 0,
+});
+
+r.on('end', common.mustCall());
+r.on('data', common.mustCall(5));