@@ -28,6 +28,7 @@ const {
2828 ObjectDefineProperty,
2929 ObjectSetPrototypeOf,
3030 SymbolAsyncIterator,
31+ Symbol
3132} = primordials ;
3233
3334module . exports = Readable ;
@@ -51,6 +52,8 @@ const {
5152 ERR_STREAM_UNSHIFT_AFTER_END_EVENT
5253} = require ( 'internal/errors' ) . codes ;
5354
55+ const kPaused = Symbol ( 'kPaused' ) ;
56+
5457// Lazy loaded to improve the startup performance.
5558let StringDecoder ;
5659let createReadableStreamAsyncIterator ;
@@ -127,7 +130,7 @@ function ReadableState(options, stream, isDuplex) {
127130 this . emittedReadable = false ;
128131 this . readableListening = false ;
129132 this . resumeScheduled = false ;
130- this . paused = true ;
133+ this [ kPaused ] = null ;
131134
132135 // Should close be emitted on destroy. Defaults to true.
133136 this . emitClose = ! options || options . emitClose !== false ;
@@ -159,6 +162,16 @@ function ReadableState(options, stream, isDuplex) {
159162 }
160163}
161164
165+ // Legacy property for `paused`
166+ ObjectDefineProperty ( ReadableState . prototype , 'paused' , {
167+ get ( ) {
168+ return this [ kPaused ] !== false ;
169+ } ,
170+ set ( value ) {
171+ this [ kPaused ] = ! ! value ;
172+ }
173+ } ) ;
174+
162175function Readable ( options ) {
163176 if ( ! ( this instanceof Readable ) )
164177 return new Readable ( options ) ;
@@ -348,7 +361,8 @@ function chunkInvalid(state, chunk) {
348361
349362
350363Readable . prototype . isPaused = function ( ) {
351- return this . _readableState . flowing === false ;
364+ const state = this . _readableState ;
365+ return state [ kPaused ] === true || state . flowing === false ;
352366} ;
353367
354368// Backwards compatibility.
@@ -947,14 +961,16 @@ function updateReadableListening(self) {
947961 const state = self . _readableState ;
948962 state . readableListening = self . listenerCount ( 'readable' ) > 0 ;
949963
950- if ( state . resumeScheduled && ! state . paused ) {
964+ if ( state . resumeScheduled && state [ kPaused ] === false ) {
951965 // Flowing needs to be set to true now, otherwise
952966 // the upcoming resume will not flow.
953967 state . flowing = true ;
954968
955969 // Crude way to check if we should resume
956970 } else if ( self . listenerCount ( 'data' ) > 0 ) {
957971 self . resume ( ) ;
972+ } else if ( ! state . readableListening ) {
973+ state . flowing = null ;
958974 }
959975}
960976
@@ -975,7 +991,7 @@ Readable.prototype.resume = function() {
975991 state . flowing = ! state . readableListening ;
976992 resume ( this , state ) ;
977993 }
978- state . paused = false ;
994+ state [ kPaused ] = false ;
979995 return this ;
980996} ;
981997
@@ -1006,7 +1022,7 @@ Readable.prototype.pause = function() {
10061022 this . _readableState . flowing = false ;
10071023 this . emit ( 'pause' ) ;
10081024 }
1009- this . _readableState . paused = true ;
1025+ this . _readableState [ kPaused ] = true ;
10101026 return this ;
10111027} ;
10121028
0 commit comments