stream: call helper function from push and unshift

PR-URL: #50173
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>

Author: rluvaton

lib/internal/streams/readable.js

@@ -284,77 +284,164 @@ Readable.prototype[SymbolAsyncDispose] = function() {
 // similar to how Writable.write() returns true if you should
 // write() some more.
 Readable.prototype.push=function(chunk,encoding){
-returnreadableAddChunk(this,chunk,encoding,false);
+debug('push',chunk);
+
+conststate=this._readableState;
+return(state[kState]&kObjectMode)===0 ?
+readableAddChunkPushByteMode(this,state,chunk,encoding) :
+readableAddChunkPushObjectMode(this,state,chunk,encoding);
 };
 
 // Unshift should *always* be something directly out of read().
 Readable.prototype.unshift=function(chunk,encoding){
-returnreadableAddChunk(this,chunk,encoding,true);
+debug('unshift',chunk);
+conststate=this._readableState;
+return(state[kState]&kObjectMode)===0 ?
+readableAddChunkUnshiftByteMode(this,state,chunk,encoding) :
+readableAddChunkUnshiftObjectMode(this,state,chunk);
 };
 
-functionreadableAddChunk(stream,chunk,encoding,addToFront){
-debug('readableAddChunk',chunk);
-conststate=stream._readableState;
 
-leterr;
-if((state[kState]&kObjectMode)===0){
-if(typeofchunk==='string'){
-encoding=encoding||state.defaultEncoding;
-if(state.encoding!==encoding){
-if(addToFront&&state.encoding){
-// When unshifting, if state.encoding is set, we have to save
-// the string in the BufferList with the state encoding.
-chunk=Buffer.from(chunk,encoding).toString(state.encoding);
-}else{
-chunk=Buffer.from(chunk,encoding);
-encoding='';
-}
+functionreadableAddChunkUnshiftByteMode(stream,state,chunk,encoding){
+if(chunk===null){
+state[kState]&=~kReading;
+onEofChunk(stream,state);
+
+returnfalse;
+}
+
+if(typeofchunk==='string'){
+encoding=encoding||state.defaultEncoding;
+if(state.encoding!==encoding){
+if(state.encoding){
+// When unshifting, if state.encoding is set, we have to save
+// the string in the BufferList with the state encoding.
+chunk=Buffer.from(chunk,encoding).toString(state.encoding);
+}else{
+chunk=Buffer.from(chunk,encoding);
 }
-}elseif(chunkinstanceofBuffer){
-encoding='';
-}elseif(Stream._isUint8Array(chunk)){
-chunk=Stream._uint8ArrayToBuffer(chunk);
-encoding='';
-}elseif(chunk!=null){
-err=newERR_INVALID_ARG_TYPE(
-'chunk',['string','Buffer','Uint8Array'],chunk);
 }
+}elseif(Stream._isUint8Array(chunk)){
+chunk=Stream._uint8ArrayToBuffer(chunk);
+}elseif(chunk!==undefined&&!(chunkinstanceofBuffer)){
+errorOrDestroy(stream,newERR_INVALID_ARG_TYPE(
+'chunk',['string','Buffer','Uint8Array'],chunk));
+returnfalse;
 }
 
-if(err){
-errorOrDestroy(stream,err);
-}elseif(chunk===null){
+
+if(!(chunk&&chunk.length>0)){
+returncanPushMore(state);
+}
+
+returnreadableAddChunkUnshiftValue(stream,state,chunk);
+}
+
+functionreadableAddChunkUnshiftObjectMode(stream,state,chunk){
+if(chunk===null){
 state[kState]&=~kReading;
 onEofChunk(stream,state);
-}elseif(((state[kState]&kObjectMode)!==0)||(chunk&&chunk.length>0)){
-if(addToFront){
-if((state[kState]&kEndEmitted)!==0)
-errorOrDestroy(stream,newERR_STREAM_UNSHIFT_AFTER_END_EVENT());
-elseif(state.destroyed||state.errored)
-returnfalse;
-else
-addChunk(stream,state,chunk,true);
-}elseif(state.ended){
-errorOrDestroy(stream,newERR_STREAM_PUSH_AFTER_EOF());
-}elseif(state.destroyed||state.errored){
-returnfalse;
-}else{
-state[kState]&=~kReading;
-if(state.decoder&&!encoding){
-chunk=state.decoder.write(chunk);
-if(state.objectMode||chunk.length!==0)
-addChunk(stream,state,chunk,false);
-else
-maybeReadMore(stream,state);
-}else{
-addChunk(stream,state,chunk,false);
-}
+
+returnfalse;
+}
+
+returnreadableAddChunkUnshiftValue(stream,state,chunk);
+}
+
+functionreadableAddChunkUnshiftValue(stream,state,chunk){
+if((state[kState]&kEndEmitted)!==0)
+errorOrDestroy(stream,newERR_STREAM_UNSHIFT_AFTER_END_EVENT());
+elseif(state.destroyed||state.errored)
+returnfalse;
+else
+addChunk(stream,state,chunk,true);
+
+returncanPushMore(state);
+}
+
+functionreadableAddChunkPushByteMode(stream,state,chunk,encoding){
+if(chunk===null){
+state[kState]&=~kReading;
+onEofChunk(stream,state);
+
+returnfalse;
+}
+
+if(typeofchunk==='string'){
+encoding=encoding||state.defaultEncoding;
+if(state.encoding!==encoding){
+chunk=Buffer.from(chunk,encoding);
+encoding='';
 }
-}elseif(!addToFront){
+}elseif(chunkinstanceofBuffer){
+encoding='';
+}elseif(Stream._isUint8Array(chunk)){
+chunk=Stream._uint8ArrayToBuffer(chunk);
+encoding='';
+}elseif(chunk!==undefined){
+errorOrDestroy(stream,newERR_INVALID_ARG_TYPE(
+'chunk',['string','Buffer','Uint8Array'],chunk));
+returnfalse;
+}
+
+if(!chunk||chunk.length<=0){
 state[kState]&=~kReading;
 maybeReadMore(stream,state);
+
+returncanPushMore(state);
+}
+
+if(state.ended){
+errorOrDestroy(stream,newERR_STREAM_PUSH_AFTER_EOF());
+
+returnfalse;
+}
+
+if(state.destroyed||state.errored){
+returnfalse;
+}
+
+state[kState]&=~kReading;
+if(state.decoder&&!encoding){
+chunk=state.decoder.write(chunk);
+if(chunk.length===0){
+maybeReadMore(stream,state);
+
+returncanPushMore(state);
+}
 }
 
+addChunk(stream,state,chunk,false);
+returncanPushMore(state);
+}
+
+functionreadableAddChunkPushObjectMode(stream,state,chunk,encoding){
+if(chunk===null){
+state[kState]&=~kReading;
+onEofChunk(stream,state);
+
+returnfalse;
+}
+
+if(state.ended){
+errorOrDestroy(stream,newERR_STREAM_PUSH_AFTER_EOF());
+returnfalse;
+}
+
+if(state.destroyed||state.errored){
+returnfalse;
+}
+
+state[kState]&=~kReading;
+if(state.decoder&&!encoding){
+chunk=state.decoder.write(chunk);
+}
+
+addChunk(stream,state,chunk,false);
+returncanPushMore(state);
+}
+
+functioncanPushMore(state){
 // We can push more data if we are below the highWaterMark.
 // Also, if we have no data yet, we can stand some more bytes.
 // This is to work around cases where hwm=0, such as the repl.