stream: 'readable' have precedence over flowing

In Streams3 the 'readable' event/.read() method had a lower precedence
than the `'data'` event that made them impossible to use them together.
This make `.resume()` a no-op if there is a listener for the
`'readable'` event, making the stream non-flowing if there is a
`'data'`  listener.

Fixes: https://github.com/nodejs/node/issues/18058

PR-URL: https://github.com/nodejs/node/pull/18994
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Anna Henningsen <anna@addaleax.net>

doc/api/stream.md

@@ -762,6 +762,9 @@ changes:
     description: >
       'readable' is always emitted in the next tick after
       .push() is called
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/18994
+    description: Using 'readable' requires calling .read().
 -->
 
 The `'readable'` event is emitted when there is data available to be read from
@@ -770,10 +773,16 @@ cause some amount of data to be read into an internal buffer.
 
 ```javascript
 const readable = getReadableStreamSomehow();
-readable.on('readable', () => {
+readable.on('readable', function() {
   // there is some data to read now
+  let data;
+
+  while (data = this.read()) {
+    console.log(data);
+  }
 });
 ```
+
 The `'readable'` event will also be emitted once the end of the stream data
 has been reached but before the `'end'` event is emitted.
 
@@ -806,6 +815,10 @@ In general, the `readable.pipe()` and `'data'` event mechanisms are easier to
 understand than the `'readable'` event. However, handling `'readable'` might
 result in increased throughput.
 
+If both `'readable'` and [`'data'`][]  are used at the same time, `'readable'`
+takes precedence in controlling the flow, i.e. `'data'` will be emitted
+only when [`stream.read()`][stream-read] is called.
+
 ##### readable.destroy([error])
 <!-- YAML
 added: v8.0.0
@@ -997,6 +1010,10 @@ the status of the `highWaterMark`.
 ##### readable.resume()
 <!-- YAML
 added: v0.9.4
+changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/18994
+    description: Resume has no effect if there is a 'readable' event listening
 -->
 
 * Returns: {this}
@@ -1016,6 +1033,9 @@ getReadableStreamSomehow()
   });
 ```
 
+The `readable.resume()` method has no effect if there is a `'readable'`
+event listener.
+
 ##### readable.setEncoding(encoding)
 <!-- YAML
 added: v0.9.4

lib/_stream_readable.js

@@ -223,6 +223,7 @@ Readable.prototype.unshift = function(chunk) {
 };
 
 function readableAddChunk(stream, chunk, encoding, addToFront, skipChunkCheck) {
+  debug('readableAddChunk', chunk);
   var state = stream._readableState;
   if (chunk === null) {
     state.reading = false;
@@ -799,20 +800,24 @@ Readable.prototype.unpipe = function(dest) {
 // Ensure readable listeners eventually get something
 Readable.prototype.on = function(ev, fn) {
   const res = Stream.prototype.on.call(this, ev, fn);
+  const state = this._readableState;
 
   if (ev === 'data') {
-    // Start flowing on next tick if stream isn't explicitly paused
-    if (this._readableState.flowing !== false)
+    // update readableListening so that resume() may be a no-op
+    // a few lines down. This is needed to support once('readable').
+    state.readableListening = this.listenerCount('readable') > 0;
+
+    // Try start flowing on next tick if stream isn't explicitly paused
+    if (state.flowing !== false)
       this.resume();
   } else if (ev === 'readable') {
-    const state = this._readableState;
     if (!state.endEmitted && !state.readableListening) {
       state.readableListening = state.needReadable = true;
       state.emittedReadable = false;
-      if (!state.reading) {
-        process.nextTick(nReadingNextTick, this);
-      } else if (state.length) {
+      if (state.length) {
         emitReadable(this);
+      } else if (!state.reading) {
+        process.nextTick(nReadingNextTick, this);
       }
     }
   }
@@ -821,6 +826,42 @@ Readable.prototype.on = function(ev, fn) {
 };
 Readable.prototype.addListener = Readable.prototype.on;
 
+Readable.prototype.removeListener = function(ev, fn) {
+  const res = Stream.prototype.removeListener.call(this, ev, fn);
+
+  if (ev === 'readable') {
+    // We need to check if there is someone still listening to
+    // to readable and reset the state. However this needs to happen
+    // after readable has been emitted but before I/O (nextTick) to
+    // support once('readable', fn) cycles. This means that calling
+    // resume within the same tick will have no
+    // effect.
+    process.nextTick(updateReadableListening, this);
+  }
+
+  return res;
+};
+
+Readable.prototype.removeAllListeners = function(ev) {
+  const res = Stream.prototype.removeAllListeners.call(this, ev);
+
+  if (ev === 'readable' || ev === undefined) {
+    // We need to check if there is someone still listening to
+    // to readable and reset the state. However this needs to happen
+    // after readable has been emitted but before I/O (nextTick) to
+    // support once('readable', fn) cycles. This means that calling
+    // resume within the same tick will have no
+    // effect.
+    process.nextTick(updateReadableListening, this);
+  }
+
+  return res;
+};
+
+function updateReadableListening(self) {
+  self._readableState.readableListening = self.listenerCount('readable') > 0;
+}
+
 function nReadingNextTick(self) {
   debug('readable nexttick read 0');
   self.read(0);
@@ -832,7 +873,9 @@ Readable.prototype.resume = function() {
   var state = this._readableState;
   if (!state.flowing) {
     debug('resume');
-    state.flowing = true;
+    // we flow only if there is no one listening
+    // for readable
+    state.flowing = !state.readableListening;
     resume(this, state);
   }
   return this;

test/parallel/test-http-readable-data-event.js

@@ -0,0 +1,52 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const http = require('http');
+const helloWorld = 'Hello World!';
+const helloAgainLater = 'Hello again later!';
+
+const server = http.createServer((req, res) => {
+  res.writeHead(200, {
+    'Content-Length': '' + (helloWorld.length + helloAgainLater.length)
+  });
+  res.write(helloWorld);
+
+  // we need to make sure the data is flushed
+  setTimeout(() => {
+    res.end(helloAgainLater);
+  }, common.platformTimeout(10));
+}).listen(0, function() {
+  const opts = {
+    hostname: 'localhost',
+    port: server.address().port,
+    path: '/'
+  };
+
+  const expectedData = [helloWorld, helloAgainLater];
+  const expectedRead = [helloWorld, null, helloAgainLater, null];
+
+  const req = http.request(opts, (res) => {
+    res.on('error', common.mustNotCall);
+
+    res.on('readable', common.mustCall(() => {
+      let data;
+
+      do {
+        data = res.read();
+        assert.strictEqual(data, expectedRead.shift());
+      } while (data !== null);
+    }, 2));
+
+    res.setEncoding('utf8');
+    res.on('data', common.mustCall((data) => {
+      assert.strictEqual(data, expectedData.shift());
+    }, 2));
+
+    res.on('end', common.mustCall(() => {
+      server.close();
+    }));
+  });
+
+  req.end();
+});

test/parallel/test-stream-readable-reading-readingMore.js

@@ -3,64 +3,166 @@ const common = require('../common');
 const assert = require('assert');
 const Readable = require('stream').Readable;
 
-const readable = new Readable({
-  read(size) {}
-});
+{
+  const readable = new Readable({
+    read(size) {}
+  });
 
-const state = readable._readableState;
+  const state = readable._readableState;
 
-// Starting off with false initially.
-assert.strictEqual(state.reading, false);
-assert.strictEqual(state.readingMore, false);
+  // Starting off with false initially.
+  assert.strictEqual(state.reading, false);
+  assert.strictEqual(state.readingMore, false);
 
-readable.on('data', common.mustCall((data) => {
-  // while in a flowing state, should try to read more.
-  if (readable.readableFlowing)
+  readable.on('data', common.mustCall((data) => {
+    // while in a flowing state with a 'readable' listener
+    // we should not be reading more
+    if (readable.readableFlowing)
+      assert.strictEqual(state.readingMore, true);
+
+    // reading as long as we've not ended
+    assert.strictEqual(state.reading, !state.ended);
+  }, 2));
+
+  function onStreamEnd() {
+    // End of stream; state.reading is false
+    // And so should be readingMore.
+    assert.strictEqual(state.readingMore, false);
+    assert.strictEqual(state.reading, false);
+  }
+
+  readable.on('readable', common.mustCall(() => {
+    // 'readable' always gets called before 'end'
+    // since 'end' hasn't been emitted, more data could be incoming
     assert.strictEqual(state.readingMore, true);
 
-  // reading as long as we've not ended
-  assert.strictEqual(state.reading, !state.ended);
-}, 2));
+    // if the stream has ended, we shouldn't be reading
+    assert.strictEqual(state.ended, !state.reading);
 
-function onStreamEnd() {
-  // End of stream; state.reading is false
-  // And so should be readingMore.
-  assert.strictEqual(state.readingMore, false);
-  assert.strictEqual(state.reading, false);
-}
+    const data = readable.read();
+    if (data === null) // reached end of stream
+      process.nextTick(common.mustCall(onStreamEnd, 1));
+  }, 2));
 
-readable.on('readable', common.mustCall(() => {
-  // 'readable' always gets called before 'end'
-  // since 'end' hasn't been emitted, more data could be incoming
+  readable.on('end', common.mustCall(onStreamEnd));
+  readable.push('pushed');
+
+  readable.read(6);
+
+  // reading
+  assert.strictEqual(state.reading, true);
   assert.strictEqual(state.readingMore, true);
 
-  // if the stream has ended, we shouldn't be reading
-  assert.strictEqual(state.ended, !state.reading);
+  // add chunk to front
+  readable.unshift('unshifted');
 
-  const data = readable.read();
-  if (data === null) // reached end of stream
-    process.nextTick(common.mustCall(onStreamEnd, 1));
-}, 2));
+  // end
+  readable.push(null);
+}
 
-readable.on('end', common.mustCall(onStreamEnd));
+{
+  const readable = new Readable({
+    read(size) {}
+  });
 
-readable.push('pushed');
+  const state = readable._readableState;
 
-// stop emitting 'data' events
-readable.pause();
+  // Starting off with false initially.
+  assert.strictEqual(state.reading, false);
+  assert.strictEqual(state.readingMore, false);
 
-// read() should only be called while operating in paused mode
-readable.read(6);
+  readable.on('data', common.mustCall((data) => {
+    // while in a flowing state without a 'readable' listener
+    // we should be reading more
+    if (readable.readableFlowing)
+      assert.strictEqual(state.readingMore, true);
 
-// reading
-assert.strictEqual(state.reading, true);
-assert.strictEqual(state.readingMore, true);
+    // reading as long as we've not ended
+    assert.strictEqual(state.reading, !state.ended);
+  }, 2));
 
-// resume emitting 'data' events
-readable.resume();
+  function onStreamEnd() {
+    // End of stream; state.reading is false
+    // And so should be readingMore.
+    assert.strictEqual(state.readingMore, false);
+    assert.strictEqual(state.reading, false);
+  }
 
-// add chunk to front
-readable.unshift('unshifted');
+  readable.on('end', common.mustCall(onStreamEnd));
+  readable.push('pushed');
 
-// end
-readable.push(null);
+  // stop emitting 'data' events
+  assert.strictEqual(state.flowing, true);
+  readable.pause();
+
+  // paused
+  assert.strictEqual(state.reading, false);
+  assert.strictEqual(state.flowing, false);
+
+  readable.resume();
+  assert.strictEqual(state.reading, false);
+  assert.strictEqual(state.flowing, true);
+
+  // add chunk to front
+  readable.unshift('unshifted');
+
+  // end
+  readable.push(null);
+}
+
+{
+  const readable = new Readable({
+    read(size) {}
+  });
+
+  const state = readable._readableState;
+
+  // Starting off with false initially.
+  assert.strictEqual(state.reading, false);
+  assert.strictEqual(state.readingMore, false);
+
+  const onReadable = common.mustNotCall;
+
+  readable.on('readable', onReadable);
+
+  readable.on('data', common.mustCall((data) => {
+    // reading as long as we've not ended
+    assert.strictEqual(state.reading, !state.ended);
+  }, 2));
+
+  readable.removeListener('readable', onReadable);
+
+  function onStreamEnd() {
+    // End of stream; state.reading is false
+    // And so should be readingMore.
+    assert.strictEqual(state.readingMore, false);
+    assert.strictEqual(state.reading, false);
+  }
+
+  readable.on('end', common.mustCall(onStreamEnd));
+  readable.push('pushed');
+
+  // we are still not flowing, we will be resuming in the next tick
+  assert.strictEqual(state.flowing, false);
+
+  // wait for nextTick, so the readableListener flag resets
+  process.nextTick(function() {
+    readable.resume();
+
+    // stop emitting 'data' events
+    assert.strictEqual(state.flowing, true);
+    readable.pause();
+
+    // paused
+    assert.strictEqual(state.flowing, false);
+
+    readable.resume();
+    assert.strictEqual(state.flowing, true);
+
+    // add chunk to front
+    readable.unshift('unshifted');
+
+    // end
+    readable.push(null);
+  });
+}

test/parallel/test-stream3-pause-then-read.js

@@ -35,6 +35,7 @@ let expectEndingData = expectTotalData;
 const r = new Readable({ highWaterMark: 1000 });
 let chunks = totalChunks;
 r._read = function(n) {
+  console.log('_read called', chunks);
   if (!(chunks % 2))
     setImmediate(push);
   else if (!(chunks % 3))
@@ -49,6 +50,7 @@ function push() {
   if (chunk) {
     totalPushed += chunk.length;
   }
+  console.log('chunks', chunks);
   r.push(chunk);
 }
 
@@ -64,6 +66,7 @@ function readn(n, then) {
   expectEndingData -= n;
   (function read() {
     const c = r.read(n);
+    console.error('c', c);
     if (!c)
       r.once('readable', read);
     else {