stream: add the pipeline destoryDestOnError option

nodejs · rickyes · Jun 30, 2020 · Jun 30, 2020 · Jun 30, 2020 · Jun 30, 2020
commit 9fc636967e1773a544671dd2f4e7e1f8ea74ff5c
diff --git a/doc/api/stream.md b/doc/api/stream.md
@@ -1627,11 +1627,15 @@ const cleanup = finished(rs, (err) => {
 });
 ```
 
-### `stream.pipeline(source[, ...transforms], destination, callback)`
+### `stream.pipeline(source[, ...transforms], destination[,options], callback)`
 ### `stream.pipeline(streams, callback)`
+
 <!-- YAML
 added: v10.0.0
 changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/
+    description: Add support for destroyDestOnError options.
   - version: v13.10.0
     pr-url: https://github.com/nodejs/node/pull/31223
     description: Add support for async generators.
@@ -1652,6 +1656,10 @@ changes:
 * `destination` {Stream|Function}
   * `source` {AsyncIterable}
   * Returns: {AsyncIterable|Promise}
+* `options` {Object}
+  * `options.destroyDestOnError` {Boolean} If value is false, the destination
+  is not destroyed when a stream error occurs.
+  **Default:** `true`.
 * `callback` {Function} Called when the pipeline is fully done.
   * `err` {Error}
   * `val` Resolved value of `Promise` returned by `destination`.

diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
@@ -25,7 +25,7 @@ let EE;
 let PassThrough;
 let createReadableStreamAsyncIterator;
 
-function destroyer(stream, reading, writing, callback) {
+function destroyer(stream, reading, writing, destroyDestOnError, callback) {
   callback = once(callback);
 
   let finished = false;
@@ -63,7 +63,12 @@ function destroyer(stream, reading, writing, callback) {
   return (err) => {
     if (finished) return;
     finished = true;
-    destroyImpl.destroyer(stream, err);
+    // When the reading is false and the writing is true, it
+    // can be considered a destination stream.
+    // If destroyDestOnError is false then stream is not destroyed.
+    if (!(reading === false && writing && destroyDestOnError === false)) {
+      destroyImpl.destroyer(stream, err);
+    }
     callback(err || new ERR_STREAM_DESTROYED('pipe'));
   };
 }
@@ -77,6 +82,13 @@ function popCallback(streams) {
   return streams.pop();
 }
 
+function popOptions(streams) {
+  const opts = streams[streams.length - 1];
+  if (!isStream(opts) &&
+      typeof opts === 'object' && opts !== null)
+    return streams.pop();
+}
+
 function isPromise(obj) {
   return !!(obj && typeof obj.then === 'function');
 }
@@ -142,13 +154,16 @@ async function pump(iterable, writable, finish) {
 
 function pipeline(...streams) {
   const callback = once(popCallback(streams));
+  const opts = popOptions(streams) || {};
 
   if (ArrayIsArray(streams[0])) streams = streams[0];
 
   if (streams.length < 2) {
     throw new ERR_MISSING_ARGS('streams');
   }
 
+  const destroyDestOnError = opts.destroyDestOnError;
+
   let error;
   let value;
   const destroys = [];
@@ -183,7 +198,8 @@ function pipeline(...streams) {
 
     if (isStream(stream)) {
       finishCount++;
-      destroys.push(destroyer(stream, reading, writing, finish));
+      destroys.push(destroyer(stream, reading, writing,
+                              destroyDestOnError, finish));
     }
 
     if (i === 0) {
@@ -241,7 +257,7 @@ function pipeline(...streams) {
         ret = pt;
 
         finishCount++;
-        destroys.push(destroyer(ret, false, true, finish));
+        destroys.push(destroyer(ret, false, true, destroyDestOnError, finish));
       }
     } else if (isStream(stream)) {
       if (isReadable(ret)) {

diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
@@ -1,5 +1,6 @@
 'use strict';
 
+const fs = require('fs');
 const common = require('../common');
 const {
   Stream,