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

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: #31036Fixes: #24474
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>

Author: ronag

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;
 
+constkPaused=Symbol('kPaused');
+
 // Lazy loaded to improve the startup performance.
 letStringDecoder;
 letcreateReadableStreamAsyncIterator;
@@ -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(){
+returnthis[kPaused]!==false;
+},
+set(value){
+this[kPaused]=!!value;
+}
+});
+
 functionReadable(options){
 if(!(thisinstanceofReadable))
 returnnewReadable(options);
@@ -348,7 +361,8 @@ function chunkInvalid(state, chunk) {
 
 
 Readable.prototype.isPaused=function(){
-returnthis._readableState.flowing===false;
+conststate=this._readableState;
+returnstate[kPaused]===true||state.flowing===false;
 };
 
 // Backwards compatibility.
@@ -947,14 +961,16 @@ function updateReadableListening(self) {
 conststate=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
 }elseif(self.listenerCount('data')>0){
 self.resume();
+}elseif(!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;
 returnthis;
 };
 
@@ -1006,7 +1022,7 @@ Readable.prototype.pause = function() {
 this._readableState.flowing=false;
 this.emit('pause');
 }
-this._readableState.paused=true;
+this._readableState[kPaused]=true;
 returnthis;
 };
 

test/parallel/test-stream-readable-data.js

@@ -0,0 +1,19 @@
+'use strict';
+constcommon=require('../common');
+
+const{ Readable }=require('stream');
+
+constreadable=newReadable({
+read(){}
+});
+
+functionread(){}
+
+readable.setEncoding('utf8');
+readable.on('readable',read);
+readable.removeListener('readable',read);
+
+process.nextTick(function(){
+readable.on('data',common.mustCall());
+readable.push('hello');
+});

test/parallel/test-stream-readable-pause-and-resume.js

@@ -1,6 +1,7 @@
 'use strict';
 
 constcommon=require('../common');
+constassert=require('assert');
 const{ Readable }=require('stream');
 
 letticks=18;
@@ -38,3 +39,20 @@ function readAndPause() {
 
 rs.on('data',ondata);
 }
+
+{
+constreadable=newReadable({
+read(){}
+});
+
+functionread(){}
+
+readable.setEncoding('utf8');
+readable.on('readable',read);
+readable.removeListener('readable',read);
+readable.pause();
+
+process.nextTick(function(){
+assert(readable.isPaused());
+});
+}