stream: pipeline with end option

Currently pipeline cannot fully replace pipe due
to the missing end option. This PR adds the end
option to the promisified pipeline method.
PR-URL: #40886
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: ronag

lib/internal/streams/pipeline.js

@@ -109,7 +109,7 @@ async function* fromReadable(val) {
 yield*Readable.prototype[SymbolAsyncIterator].call(val);
 }
 
-asyncfunctionpump(iterable,writable,finish){
+asyncfunctionpump(iterable,writable,finish,opts){
 leterror;
 letonresolve=null;
 
@@ -153,7 +153,9 @@ async function pump(iterable, writable, finish) {
 }
 }
 
-writable.end();
+if(opts?.end!==false){
+writable.end();
+}
 
 awaitwait();
 
@@ -227,17 +229,22 @@ function pipelineImpl(streams, callback, opts) {
 conststream=streams[i];
 constreading=i<streams.length-1;
 constwriting=i>0;
+constend=reading||opts?.end!==false;
 
 if(isNodeStream(stream)){
-finishCount++;
-destroys.push(destroyer(stream,reading,writing,(err)=>{
-if(!err&&!reading&&isReadableFinished(stream,false)){
-stream.read(0);
-destroyer(stream,true,writing,finish);
-}else{
-finish(err);
-}
-}));
+if(end){
+finishCount++;
+destroys.push(destroyer(stream,reading,writing,(err)=>{
+if(!err&&!reading&&isReadableFinished(stream,false)){
+stream.read(0);
+destroyer(stream,true,writing,finish);
+}else{
+finish(err);
+}
+}));
+}else{
+stream.on('error',finish);
+}
 }
 
 if(i===0){
@@ -282,14 +289,17 @@ function pipelineImpl(streams, callback, opts) {
 then.call(ret,
 (val)=>{
 value=val;
-pt.end(val);
+pt.write(val);
+if(end){
+pt.end();
+}
 },(err)=>{
 pt.destroy(err);
 },
 );
 }elseif(isIterable(ret,true)){
 finishCount++;
-pump(ret,pt,finish);
+pump(ret,pt,finish,{ end });
 }else{
 thrownewERR_INVALID_RETURN_VALUE(
 'AsyncIterable or Promise','destination',ret);
@@ -302,7 +312,7 @@ function pipelineImpl(streams, callback, opts) {
 }
 }elseif(isNodeStream(stream)){
 if(isReadableNodeStream(ret)){
-ret.pipe(stream);
+ret.pipe(stream,{ end });
 
 // Compat. Before node v10.12.0 stdio used to throw an error so
 // pipe() did/does not end() stdio destinations.
@@ -314,7 +324,7 @@ function pipelineImpl(streams, callback, opts) {
 ret=makeAsyncIterable(ret);
 
 finishCount++;
-pump(ret,stream,finish);
+pump(ret,stream,finish,{ end });
 }
 ret=stream;
 }else{

lib/stream/promises.js

@@ -16,11 +16,13 @@ const eos = require('internal/streams/end-of-stream');
 functionpipeline(...streams){
 returnnewPromise((resolve,reject)=>{
 letsignal;
+letend;
 constlastArg=streams[streams.length-1];
 if(lastArg&&typeoflastArg==='object'&&
 !isNodeStream(lastArg)&&!isIterable(lastArg)){
 constoptions=ArrayPrototypePop(streams);
 signal=options.signal;
+end=options.end;
 }
 
 pl(streams,(err,value)=>{
@@ -29,7 +31,7 @@ function pipeline(...streams) {
 }else{
 resolve(value);
 }
-},{ signal });
+},{ signal, end});
 });
 }
 

test/parallel/test-stream-pipeline.js

@@ -1465,5 +1465,26 @@ const tsp = require('timers/promises');
 assert.strictEqual(duplex.destroyed,true);
 }
 
-run();
+run().then(common.mustCall());
+}
+
+{
+constpipelinePromise=promisify(pipeline);
+
+asyncfunctionrun(){
+constread=newReadable({
+read(){}
+});
+
+constduplex=newPassThrough();
+
+read.push(null);
+
+awaitpipelinePromise(read,duplex,{end: false});
+
+assert.strictEqual(duplex.destroyed,false);
+assert.strictEqual(duplex.writableEnded,false);
+}
+
+run().then(common.mustCall());
 }