stream: reset flowing state if no 'readable' or 'data' listeners

ronag · targos · commit 84e8e1a1384c · 2020-01-14T14:30:39.000+01:00
If we don't have any 'readable' or 'data' listeners and we are not about to resume. Then reset flowing state to initial null state. PR-URL: #31036 Fixes: #24474 Reviewed-By: Luigi Pinca <luigipinca@gmail.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Rich Trott <rtrott@gmail.com>
diff --git a/lib/_stream_readable.js b/lib/_stream_readable.js
@@ -28,6 +28,7 @@ const {
   ObjectDefineProperty,
   ObjectSetPrototypeOf,
   SymbolAsyncIterator,
+  Symbol
 } = primordials;
 
 module.exports = Readable;
@@ -51,6 +52,8 @@ const {
   ERR_STREAM_UNSHIFT_AFTER_END_EVENT
 } = require('internal/errors').codes;
 
+const kPaused = Symbol('kPaused');
+
 // Lazy loaded to improve the startup performance.
 let StringDecoder;
 let createReadableStreamAsyncIterator;
@@ -127,7 +130,7 @@ function ReadableState(options, stream, isDuplex) {
   this.emittedReadable = false;
   this.readableListening = false;
   this.resumeScheduled = false;
-  this.paused = true;
+  this[kPaused] = null;
 
   // Should close be emitted on destroy. Defaults to true.
   this.emitClose = !options || options.emitClose !== false;
@@ -159,6 +162,16 @@ function ReadableState(options, stream, isDuplex) {
   }
 }
 
+// Legacy property for `paused`
+ObjectDefineProperty(ReadableState.prototype, 'paused', {
+  get() {
+    return this[kPaused] !== false;
+  },
+  set(value) {
+    this[kPaused] = !!value;
+  }
+});
+
 function Readable(options) {
   if (!(this instanceof Readable))
     return new Readable(options);
@@ -348,7 +361,8 @@ function chunkInvalid(state, chunk) {
 
 
 Readable.prototype.isPaused = function() {
-  return this._readableState.flowing === false;
+  const state = this._readableState;
+  return state[kPaused] === true || state.flowing === false;
 };
 
 // Backwards compatibility.
@@ -947,14 +961,16 @@ function updateReadableListening(self) {
   const state = self._readableState;
   state.readableListening = self.listenerCount('readable') > 0;
 
-  if (state.resumeScheduled && !state.paused) {
+  if (state.resumeScheduled && state[kPaused] === false) {
     // Flowing needs to be set to true now, otherwise
     // the upcoming resume will not flow.
     state.flowing = true;
 
     // Crude way to check if we should resume
   } else if (self.listenerCount('data') > 0) {
     self.resume();
+  } else if (!state.readableListening) {
+    state.flowing = null;
   }
 }
 
@@ -975,7 +991,7 @@ Readable.prototype.resume = function() {
     state.flowing = !state.readableListening;
     resume(this, state);
   }
-  state.paused = false;
+  state[kPaused] = false;
   return this;
 };
 
@@ -1006,7 +1022,7 @@ Readable.prototype.pause = function() {
     this._readableState.flowing = false;
     this.emit('pause');
   }
-  this._readableState.paused = true;
+  this._readableState[kPaused] = true;
   return this;
 };
 
diff --git a/test/parallel/test-stream-readable-data.js b/test/parallel/test-stream-readable-data.js
@@ -0,0 +1,19 @@
+'use strict';
+const common = require('../common');
+
+const { Readable } = require('stream');
+
+const readable = new Readable({
+  read() {}
+});
+
+function read() {}
+
+readable.setEncoding('utf8');
+readable.on('readable', read);
+readable.removeListener('readable', read);
+
+process.nextTick(function() {
+  readable.on('data', common.mustCall());
+  readable.push('hello');
+});
diff --git a/test/parallel/test-stream-readable-pause-and-resume.js b/test/parallel/test-stream-readable-pause-and-resume.js
@@ -1,6 +1,7 @@
 'use strict';
 
 const common = require('../common');
+const assert = require('assert');
 const { Readable } = require('stream');
 
 let ticks = 18;
@@ -38,3 +39,20 @@ function readAndPause() {
 
   rs.on('data', ondata);
 }
+
+{
+  const readable = new Readable({
+    read() {}
+  });
+
+  function read() {}
+
+  readable.setEncoding('utf8');
+  readable.on('readable', read);
+  readable.removeListener('readable', read);
+  readable.pause();
+
+  process.nextTick(function() {
+    assert(readable.isPaused());
+  });
+}
