@@ -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) |
|