stream: don't destroy final readable stream in pipeline

If the last stream in a pipeline is still usable/readable don't destroy it to allow further composition. Fixes: #32105 Backport-PR-URL: #32111 PR-URL: #32110 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Luigi Pinca <[email protected]>

Author: ronag

lib/internal/streams/pipeline.js

@@ -36,15 +36,18 @@ function destroyStream(stream, err){
 if(typeofstream.close==='function')returnstream.close();
 }
 
-functiondestroyer(stream,reading,writing,callback){
+functiondestroyer(stream,reading,writing,final,callback){
 callback=once(callback);
 letdestroyed=false;
 
 if(eos===undefined)eos=require('internal/streams/end-of-stream');
 eos(stream,{readable: reading,writable: writing},(err)=>{
 if(destroyed)return;
 destroyed=true;
-destroyStream(stream,err);
+constreadable=stream.readable||isRequest(stream);
+if(err||!final||!readable){
+destroyStream(stream,err);
+}
 callback(err);
 });
 
@@ -176,7 +179,7 @@ function pipeline(...streams){
 }
 
 functionwrap(stream,reading,writing,final){
-destroys.push(destroyer(stream,reading,writing,(err)=>{
+destroys.push(destroyer(stream,reading,writing,final,(err)=>{
 finish(err,final);
 }));
 }

test/parallel/test-stream-pipeline.js

@@ -918,6 +918,52 @@ const{promisify } = require('util');
 constdst=newPassThrough({autoDestroy: false});
 pipeline(src,dst,common.mustCall(()=>{
 assert.strictEqual(src.destroyed,true);
+assert.strictEqual(dst.destroyed,false);
+}));
+src.end();
+}
+
+{
+constserver=http.createServer((req,res)=>{
+});
+
+server.listen(0,()=>{
+constreq=http.request({
+port: server.address().port
+});
+
+constbody=newPassThrough();
+pipeline(
+body,
+req,
+common.mustCall((err)=>{
+assert(!err);
+assert(!req.res);
+assert(!req.aborted);
+req.abort();
+server.close();
+})
+);
+body.end();
+});
+}
+
+{
+constsrc=newPassThrough();
+constdst=newPassThrough();
+pipeline(src,dst,common.mustCall((err)=>{
+assert(!err);
+assert.strictEqual(dst.destroyed,false);
+}));
+src.end();
+}
+
+{
+constsrc=newPassThrough();
+constdst=newPassThrough();
+dst.readable=false;
+pipeline(src,dst,common.mustCall((err)=>{
+assert(!err);
 assert.strictEqual(dst.destroyed,true);
 }));
 src.end();