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

PR-URL: #32110
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>

lib/internal/streams/pipeline.js

@@ -25,9 +25,12 @@ let EE;
 let PassThrough;
 let createReadableStreamAsyncIterator;
 
-function destroyer(stream, reading, writing, callback) {
+function destroyer(stream, reading, writing, final, callback) {
   const _destroy = once((err) => {
-    destroyImpl.destroyer(stream, err);
+    const readable = stream.readable || isRequest(stream);
+    if (err || !final || !readable) {
+      destroyImpl.destroyer(stream, err);
+    }
     callback(err);
   });
 
@@ -68,6 +71,10 @@ function popCallback(streams) {
   return streams.pop();
 }
 
+function isRequest(stream) {
+  return stream.setHeader && typeof stream.abort === 'function';
+}
+
 function isPromise(obj) {
   return !!(obj && typeof obj.then === 'function');
 }
@@ -159,7 +166,7 @@ function pipeline(...streams) {
   }
 
   function wrap(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

@@ -916,7 +916,7 @@ const { promisify } = require('util');
   const dst = new PassThrough({ autoDestroy: false });
   pipeline(src, dst, common.mustCall(() => {
     assert.strictEqual(src.destroyed, true);
-    assert.strictEqual(dst.destroyed, true);
+    assert.strictEqual(dst.destroyed, false);
   }));
   src.end();
 }
@@ -938,3 +938,49 @@ const { promisify } = require('util');
   r.push(null);
   r.emit('close');
 }
+
+{
+  const server = http.createServer((req, res) => {
+  });
+
+  server.listen(0, () => {
+    const req = http.request({
+      port: server.address().port
+    });
+
+    const body = new PassThrough();
+    pipeline(
+      body,
+      req,
+      common.mustCall((err) => {
+        assert(!err);
+        assert(!req.res);
+        assert(!req.aborted);
+        req.abort();
+        server.close();
+      })
+    );
+    body.end();
+  });
+}
+
+{
+  const src = new PassThrough();
+  const dst = new PassThrough();
+  pipeline(src, dst, common.mustCall((err) => {
+    assert(!err);
+    assert.strictEqual(dst.destroyed, false);
+  }));
+  src.end();
+}
+
+{
+  const src = new PassThrough();
+  const dst = new PassThrough();
+  dst.readable = false;
+  pipeline(src, dst, common.mustCall((err) => {
+    assert(!err);
+    assert.strictEqual(dst.destroyed, true);
+  }));
+  src.end();
+}