@@ -82,6 +82,75 @@ const nop = () => {};
82
82
83
83
const { errorOrDestroy } = destroyImpl ;
84
84
85
+ const kObjectMode = 1 << 0 ;
86
+ const kEnded = 1 << 1 ;
87
+ const kEndEmitted = 1 << 2 ;
88
+ const kReading = 1 << 3 ;
89
+ const kConstructed = 1 << 4 ;
90
+ const kSync = 1 << 5 ;
91
+ const kNeedReadable = 1 << 6 ;
92
+ const kEmittedReadable = 1 << 7 ;
93
+ const kReadableListening = 1 << 8 ;
94
+ const kResumeScheduled = 1 << 9 ;
95
+ const kErrorEmitted = 1 << 10 ;
96
+ const kEmitClose = 1 << 11 ;
97
+ const kAutoDestroy = 1 << 12 ;
98
+ const kDestroyed = 1 << 13 ;
99
+ const kClosed = 1 << 14 ;
100
+ const kCloseEmitted = 1 << 15 ;
101
+ const kMultiAwaitDrain = 1 << 16 ;
102
+ const kReadingMore = 1 << 17 ;
103
+ const kDataEmitted = 1 << 18 ;
104
+
105
+ // TODO(benjamingr) it is likely slower to do it this way than with free functions
106
+ function makeBitMapDescriptor ( bit ) {
107
+ return {
108
+ enumerable : false ,
109
+ get ( ) { return ( this . state & bit ) !== 0 ; } ,
110
+ set ( value ) {
111
+ if ( value ) this . state |= bit ;
112
+ else this . state &= ~ bit ;
113
+ } ,
114
+ } ;
115
+ }
116
+ ObjectDefineProperties ( ReadableState . prototype , {
117
+ objectMode : makeBitMapDescriptor ( kObjectMode ) ,
118
+ ended : makeBitMapDescriptor ( kEnded ) ,
119
+ endEmitted : makeBitMapDescriptor ( kEndEmitted ) ,
120
+ reading : makeBitMapDescriptor ( kReading ) ,
121
+ // Stream is still being constructed and cannot be
122
+ // destroyed until construction finished or failed.
123
+ // Async construction is opt in, therefore we start as
124
+ // constructed.
125
+ constructed : makeBitMapDescriptor ( kConstructed ) ,
126
+ // A flag to be able to tell if the event 'readable'/'data' is emitted
127
+ // immediately, or on a later tick. We set this to true at first, because
128
+ // any actions that shouldn't happen until "later" should generally also
129
+ // not happen before the first read call.
130
+ sync : makeBitMapDescriptor ( kSync ) ,
131
+ // Whenever we return null, then we set a flag to say
132
+ // that we're awaiting a 'readable' event emission.
133
+ needReadable : makeBitMapDescriptor ( kNeedReadable ) ,
134
+ emittedReadable : makeBitMapDescriptor ( kEmittedReadable ) ,
135
+ readableListening : makeBitMapDescriptor ( kReadableListening ) ,
136
+ resumeScheduled : makeBitMapDescriptor ( kResumeScheduled ) ,
137
+ // True if the error was already emitted and should not be thrown again.
138
+ errorEmitted : makeBitMapDescriptor ( kErrorEmitted ) ,
139
+ emitClose : makeBitMapDescriptor ( kEmitClose ) ,
140
+ autoDestroy : makeBitMapDescriptor ( kAutoDestroy ) ,
141
+ // Has it been destroyed.
142
+ destroyed : makeBitMapDescriptor ( kDestroyed ) ,
143
+ // Indicates whether the stream has finished destroying.
144
+ closed : makeBitMapDescriptor ( kClosed ) ,
145
+ // True if close has been emitted or would have been emitted
146
+ // depending on emitClose.
147
+ closeEmitted : makeBitMapDescriptor ( kCloseEmitted ) ,
148
+ multiAwaitDrain : makeBitMapDescriptor ( kMultiAwaitDrain ) ,
149
+ // If true, a maybeReadMore has been scheduled.
150
+ readingMore : makeBitMapDescriptor ( kReadingMore ) ,
151
+ dataEmitted : makeBitMapDescriptor ( kDataEmitted ) ,
152
+ } ) ;
153
+
85
154
function ReadableState ( options , stream , isDuplex ) {
86
155
// Duplex streams are both readable and writable, but share
87
156
// the same options object.
@@ -91,13 +160,15 @@ function ReadableState(options, stream, isDuplex) {
91
160
if ( typeof isDuplex !== 'boolean' )
92
161
isDuplex = stream instanceof Stream . Duplex ;
93
162
163
+ // Bit map field to store ReadableState more effciently with 1 bit per field
164
+ // instead of a V8 slot per field.
165
+ this . state = kEmitClose | kAutoDestroy | kConstructed | kSync ;
94
166
// Object stream flag. Used to make read(n) ignore n and to
95
167
// make all the buffer merging and length checks go away.
96
- this . objectMode = ! ! ( options && options . objectMode ) ;
168
+ if ( options && options . objectMode ) this . state |= kObjectMode ;
97
169
98
- if ( isDuplex )
99
- this . objectMode = this . objectMode ||
100
- ! ! ( options && options . readableObjectMode ) ;
170
+ if ( isDuplex && options && options . readableObjectMode )
171
+ this . state |= kObjectMode ;
101
172
102
173
// The point at which it stops calling _read() to fill the buffer
103
174
// Note: 0 is a valid value, means "don't call _read preemptively ever"
@@ -112,54 +183,22 @@ function ReadableState(options, stream, isDuplex) {
112
183
this . length = 0 ;
113
184
this . pipes = [ ] ;
114
185
this . flowing = null ;
115
- this . ended = false ;
116
- this . endEmitted = false ;
117
- this . reading = false ;
118
-
119
- // Stream is still being constructed and cannot be
120
- // destroyed until construction finished or failed.
121
- // Async construction is opt in, therefore we start as
122
- // constructed.
123
- this . constructed = true ;
124
186
125
- // A flag to be able to tell if the event 'readable'/'data' is emitted
126
- // immediately, or on a later tick. We set this to true at first, because
127
- // any actions that shouldn't happen until "later" should generally also
128
- // not happen before the first read call.
129
- this . sync = true ;
130
-
131
- // Whenever we return null, then we set a flag to say
132
- // that we're awaiting a 'readable' event emission.
133
- this . needReadable = false ;
134
- this . emittedReadable = false ;
135
- this . readableListening = false ;
136
- this . resumeScheduled = false ;
137
187
this [ kPaused ] = null ;
138
188
139
- // True if the error was already emitted and should not be thrown again.
140
- this . errorEmitted = false ;
141
-
142
189
// Should close be emitted on destroy. Defaults to true.
143
- this . emitClose = ! options || options . emitClose !== false ;
190
+ if ( options && options . emitClose === false ) this . state &= ~ kEmitClose ;
144
191
145
192
// Should .destroy() be called after 'end' (and potentially 'finish').
146
- this . autoDestroy = ! options || options . autoDestroy !== false ;
193
+ if ( options && options . autoDestroy === false ) this . state &= ~ kAutoDestroy ;
147
194
148
- // Has it been destroyed.
149
- this . destroyed = false ;
150
195
151
196
// Indicates whether the stream has errored. When true no further
152
197
// _read calls, 'data' or 'readable' events should occur. This is needed
153
198
// since when autoDestroy is disabled we need a way to tell whether the
154
199
// stream has failed.
155
200
this . errored = null ;
156
201
157
- // Indicates whether the stream has finished destroying.
158
- this . closed = false ;
159
-
160
- // True if close has been emitted or would have been emitted
161
- // depending on emitClose.
162
- this . closeEmitted = false ;
163
202
164
203
// Crypto is kind of old and crusty. Historically, its default string
165
204
// encoding is 'binary' so we have to make this configurable.
@@ -169,12 +208,6 @@ function ReadableState(options, stream, isDuplex) {
169
208
// Ref the piped dest which we need a drain event on it
170
209
// type: null | Writable | Set<Writable>.
171
210
this . awaitDrainWriters = null ;
172
- this . multiAwaitDrain = false ;
173
-
174
- // If true, a maybeReadMore has been scheduled.
175
- this . readingMore = false ;
176
-
177
- this . dataEmitted = false ;
178
211
179
212
this . decoder = null ;
180
213
this . encoding = null ;
@@ -255,7 +288,7 @@ function readableAddChunk(stream, chunk, encoding, addToFront) {
255
288
const state = stream . _readableState ;
256
289
257
290
let err ;
258
- if ( ! state . objectMode ) {
291
+ if ( ( state . state & kObjectMode ) === 0 ) {
259
292
if ( typeof chunk === 'string' ) {
260
293
encoding = encoding || state . defaultEncoding ;
261
294
if ( state . encoding !== encoding ) {
@@ -282,11 +315,11 @@ function readableAddChunk(stream, chunk, encoding, addToFront) {
282
315
if ( err ) {
283
316
errorOrDestroy ( stream , err ) ;
284
317
} else if ( chunk === null ) {
285
- state . reading = false ;
318
+ state . state &= ~ kReading ;
286
319
onEofChunk ( stream , state ) ;
287
- } else if ( state . objectMode || ( chunk && chunk . length > 0 ) ) {
320
+ } else if ( ( ( state . state & kObjectMode ) !== 0 ) || ( chunk && chunk . length > 0 ) ) {
288
321
if ( addToFront ) {
289
- if ( state . endEmitted )
322
+ if ( ( state . state & kEndEmitted ) !== 0 )
290
323
errorOrDestroy ( stream , new ERR_STREAM_UNSHIFT_AFTER_END_EVENT ( ) ) ;
291
324
else if ( state . destroyed || state . errored )
292
325
return false ;
@@ -297,7 +330,7 @@ function readableAddChunk(stream, chunk, encoding, addToFront) {
297
330
} else if ( state . destroyed || state . errored ) {
298
331
return false ;
299
332
} else {
300
- state . reading = false ;
333
+ state . state &= ~ kReading ;
301
334
if ( state . decoder && ! encoding ) {
302
335
chunk = state . decoder . write ( chunk ) ;
303
336
if ( state . objectMode || chunk . length !== 0 )
@@ -309,7 +342,7 @@ function readableAddChunk(stream, chunk, encoding, addToFront) {
309
342
}
310
343
}
311
344
} else if ( ! addToFront ) {
312
- state . reading = false ;
345
+ state . state &= ~ kReading ;
313
346
maybeReadMore ( stream , state ) ;
314
347
}
315
348
@@ -325,7 +358,7 @@ function addChunk(stream, state, chunk, addToFront) {
325
358
stream . listenerCount ( 'data' ) > 0 ) {
326
359
// Use the guard to avoid creating `Set()` repeatedly
327
360
// when we have multiple pipes.
328
- if ( state . multiAwaitDrain ) {
361
+ if ( ( state . state & kMultiAwaitDrain ) !== 0 ) {
329
362
state . awaitDrainWriters . clear ( ) ;
330
363
} else {
331
364
state . awaitDrainWriters = null ;
@@ -341,7 +374,7 @@ function addChunk(stream, state, chunk, addToFront) {
341
374
else
342
375
state . buffer . push ( chunk ) ;
343
376
344
- if ( state . needReadable )
377
+ if ( ( state . state & kNeedReadable ) !== 0 )
345
378
emitReadable ( stream ) ;
346
379
}
347
380
maybeReadMore ( stream , state ) ;
@@ -396,7 +429,7 @@ function computeNewHighWaterMark(n) {
396
429
function howMuchToRead ( n , state ) {
397
430
if ( n <= 0 || ( state . length === 0 && state . ended ) )
398
431
return 0 ;
399
- if ( state . objectMode )
432
+ if ( ( state . state & kObjectMode ) !== 0 )
400
433
return 1 ;
401
434
if ( NumberIsNaN ( n ) ) {
402
435
// Only flow one buffer at a time.
@@ -427,7 +460,7 @@ Readable.prototype.read = function(n) {
427
460
state . highWaterMark = computeNewHighWaterMark ( n ) ;
428
461
429
462
if ( n !== 0 )
430
- state . emittedReadable = false ;
463
+ state . state &= ~ kEmittedReadable ;
431
464
432
465
// If we're doing read(0) to trigger a readable event, but we
433
466
// already have a bunch of data in the buffer, then just trigger
@@ -478,7 +511,7 @@ Readable.prototype.read = function(n) {
478
511
// 3. Actually pull the requested chunks out of the buffer and return.
479
512
480
513
// if we need a readable event, then we need to do some reading.
481
- let doRead = state . needReadable ;
514
+ let doRead = ( state . state & kNeedReadable ) !== 0 ;
482
515
debug ( 'need readable' , doRead ) ;
483
516
484
517
// If we currently have less than the highWaterMark, then also read some.
@@ -496,20 +529,19 @@ Readable.prototype.read = function(n) {
496
529
debug ( 'reading, ended or constructing' , doRead ) ;
497
530
} else if ( doRead ) {
498
531
debug ( 'do read' ) ;
499
- state . reading = true ;
500
- state . sync = true ;
532
+ state . state |= kReading | kSync ;
501
533
// If the length is currently zero, then we *need* a readable event.
502
534
if ( state . length === 0 )
503
- state . needReadable = true ;
535
+ state . state |= kNeedReadable ;
504
536
505
537
// Call internal read method
506
538
try {
507
539
this . _read ( state . highWaterMark ) ;
508
540
} catch ( err ) {
509
541
errorOrDestroy ( this , err ) ;
510
542
}
543
+ state . state &= ~ kSync ;
511
544
512
- state . sync = false ;
513
545
// If _read pushed data synchronously, then `reading` will be false,
514
546
// and we need to re-evaluate how much data we can return to the user.
515
547
if ( ! state . reading )
0 commit comments