Skip to content

Commit a3cb8ba

Browse files
committed
stream: consolidate common code from push and unshift helper functions
1 parent 31e0759 commit a3cb8ba

File tree

1 file changed

+25
-54
lines changed

1 file changed

+25
-54
lines changed

lib/internal/streams/readable.js

Lines changed: 25 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -367,6 +367,18 @@ Readable.prototype.push = function(chunk, encoding) {
367367
debug('push', chunk);
368368

369369
const state = this._readableState;
370+
if ((state[kState] & kEnded) !== 0) {
371+
errorOrDestroy(this, new ERR_STREAM_PUSH_AFTER_EOF());
372+
return false;
373+
} else if ((state[kState] & (kDestroyed | kErrored)) !== 0)
374+
return false;
375+
376+
state[kState] &= ~kReading;
377+
if (chunk === null) {
378+
onEofChunk(this, state);
379+
return false;
380+
}
381+
370382
return (state[kState] & kObjectMode) === 0 ?
371383
readableAddChunkPushByteMode(this, state, chunk, encoding) :
372384
readableAddChunkPushObjectMode(this, state, chunk, encoding);
@@ -376,20 +388,25 @@ Readable.prototype.push = function(chunk, encoding) {
376388
Readable.prototype.unshift = function(chunk, encoding) {
377389
debug('unshift', chunk);
378390
const state = this._readableState;
391+
if ((state[kState] & kEndEmitted) !== 0) {
392+
errorOrDestroy(this, new ERR_STREAM_UNSHIFT_AFTER_END_EVENT());
393+
return false;
394+
} else if ((state[kState] & (kDestroyed | kErrored)) !== 0)
395+
return false;
396+
397+
if (chunk === null) {
398+
state[kState] &= ~kReading;
399+
onEofChunk(this, state);
400+
return false;
401+
}
402+
379403
return (state[kState] & kObjectMode) === 0 ?
380404
readableAddChunkUnshiftByteMode(this, state, chunk, encoding) :
381405
readableAddChunkUnshiftObjectMode(this, state, chunk);
382406
};
383407

384408

385409
function readableAddChunkUnshiftByteMode(stream, state, chunk, encoding) {
386-
if (chunk === null) {
387-
state[kState] &= ~kReading;
388-
onEofChunk(stream, state);
389-
390-
return false;
391-
}
392-
393410
if (typeof chunk === 'string') {
394411
encoding = encoding || state.defaultEncoding;
395412
if (state.encoding !== encoding) {
@@ -418,34 +435,16 @@ function readableAddChunkUnshiftByteMode(stream, state, chunk, encoding) {
418435
}
419436

420437
function readableAddChunkUnshiftObjectMode(stream, state, chunk) {
421-
if (chunk === null) {
422-
state[kState] &= ~kReading;
423-
onEofChunk(stream, state);
424-
425-
return false;
426-
}
427-
428438
return readableAddChunkUnshiftValue(stream, state, chunk);
429439
}
430440

431441
function readableAddChunkUnshiftValue(stream, state, chunk) {
432-
if ((state[kState] & kEndEmitted) !== 0)
433-
errorOrDestroy(stream, new ERR_STREAM_UNSHIFT_AFTER_END_EVENT());
434-
else if ((state[kState] & (kDestroyed | kErrored)) !== 0)
435-
return false;
436-
else
437-
addChunk(stream, state, chunk, true);
442+
addChunk(stream, state, chunk, true);
438443

439444
return canPushMore(state);
440445
}
441446

442447
function readableAddChunkPushByteMode(stream, state, chunk, encoding) {
443-
if (chunk === null) {
444-
state[kState] &= ~kReading;
445-
onEofChunk(stream, state);
446-
return false;
447-
}
448-
449448
if (typeof chunk === 'string') {
450449
encoding = encoding || state.defaultEncoding;
451450
if (state.encoding !== encoding) {
@@ -464,22 +463,11 @@ function readableAddChunkPushByteMode(stream, state, chunk, encoding) {
464463
}
465464

466465
if (!chunk || chunk.length <= 0) {
467-
state[kState] &= ~kReading;
468466
maybeReadMore(stream, state);
469467

470468
return canPushMore(state);
471469
}
472470

473-
if ((state[kState] & kEnded) !== 0) {
474-
errorOrDestroy(stream, new ERR_STREAM_PUSH_AFTER_EOF());
475-
return false;
476-
}
477-
478-
if ((state[kState] & (kDestroyed | kErrored)) !== 0) {
479-
return false;
480-
}
481-
482-
state[kState] &= ~kReading;
483471
if ((state[kState] & kDecoder) !== 0 && !encoding) {
484472
chunk = state[kDecoderValue].write(chunk);
485473
if (chunk.length === 0) {
@@ -493,23 +481,6 @@ function readableAddChunkPushByteMode(stream, state, chunk, encoding) {
493481
}
494482

495483
function readableAddChunkPushObjectMode(stream, state, chunk, encoding) {
496-
if (chunk === null) {
497-
state[kState] &= ~kReading;
498-
onEofChunk(stream, state);
499-
return false;
500-
}
501-
502-
if ((state[kState] & kEnded) !== 0) {
503-
errorOrDestroy(stream, new ERR_STREAM_PUSH_AFTER_EOF());
504-
return false;
505-
}
506-
507-
if ((state[kState] & (kDestroyed | kErrored)) !== 0) {
508-
return false;
509-
}
510-
511-
state[kState] &= ~kReading;
512-
513484
if ((state[kState] & kDecoder) !== 0 && !encoding) {
514485
chunk = state[kDecoderValue].write(chunk);
515486
}

0 commit comments

Comments
 (0)