stream: restore flow if there are 'data' handlers after once('readable')

Fixes: https://github.com/nodejs/node/issues/21398
See: https://github.com/nodejs/node/pull/21696

PR-URL: https://github.com/nodejs/node/pull/22209
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Mathias Buus <mathiasbuus@gmail.com>

1 files changed, 8 insertions, 1 deletions

diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
index 029faaa614..f082996643 100644
--- a/lib/_stream_readable.js
+++ b/lib/_stream_readable.js
@@ -810,6 +810,7 @@ Readable.prototype.on = function(ev, fn) {
   } else if (ev === 'readable') {
     if (!state.endEmitted && !state.readableListening) {
       state.readableListening = state.needReadable = true;
+      state.flowing = false;
       state.emittedReadable = false;
       debug('on readable', state.length, state.reading);
       if (state.length) {
@@ -858,6 +859,11 @@ Readable.prototype.removeAllListeners = function(ev) {
 
 function updateReadableListening(self) {
   self._readableState.readableListening = self.listenerCount('readable') > 0;
+
+  // crude way to check if we should resume
+  if (self.listenerCount('data') > 0) {
+    self.resume();
+  }
 }
 
 function nReadingNextTick(self) {
@@ -872,7 +878,8 @@ Readable.prototype.resume = function() {
   if (!state.flowing) {
     debug('resume');
     // we flow only if there is no one listening
-    // for readable
+    // for readable, but we still have to call
+    // resume()
     state.flowing = !state.readableListening;
     resume(this, state);
   }