streams: pipeline with signal

ronag · ronag · commit 8578ba0a1c03 · 2021-06-18T07:36:51.000+02:00
Generators in pipeline must be able to be aborted or pipeline
can deadlock.
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
@@ -22,6 +22,7 @@ const {
     ERR_STREAM_DESTROYED,
     ERR_STREAM_PREMATURE_CLOSE,
   },
+  AbortError,
 } = require('internal/errors');
 
 const { validateCallback } = require('internal/validators');
@@ -34,6 +35,7 @@ const {
   isStream,
 } = require('internal/streams/utils');
 const assert = require('internal/assert');
+const { AbortController } = require('internal/abort_controller');
 
 let PassThrough;
 let Readable;
@@ -176,10 +178,24 @@ function pipeline(...streams) {
     streams = streams[0];
   }
 
+  return pipelineImpl(streams, callback);
+}
+
+function pipelineImpl(streams, callback, opts) {
   if (streams.length < 2) {
     throw new ERR_MISSING_ARGS('streams');
   }
 
+  const ac = new AbortController();
+  const signal = ac.signal;
+  const outerSignal = opts?.signal;
+
+  function abort () {
+    finishImpl(new AbortError());
+  }
+
+  outerSignal?.addEventListener('abort', abort);
+
   let error;
   let value;
   const destroys = [];
@@ -188,7 +204,10 @@ function pipeline(...streams) {
 
   function finish(err) {
     const final = --finishCount === 0;
+    finishImpl(err, final);
+  }
 
+  function finishImpl(err, final) {
     if (err && (!error || error.code === 'ERR_STREAM_PREMATURE_CLOSE')) {
       error = err;
     }
@@ -201,6 +220,9 @@ function pipeline(...streams) {
       destroys.shift()(error);
     }
 
+    outerSignal?.removeEventListener('abort', abort);
+    ac.abort();
+
     if (final) {
       callback(error, value);
     }
@@ -219,7 +241,7 @@ function pipeline(...streams) {
 
     if (i === 0) {
       if (typeof stream === 'function') {
-        ret = stream();
+        ret = stream(signal);
         if (!isIterable(ret)) {
           throw new ERR_INVALID_RETURN_VALUE(
             'Iterable, AsyncIterable or Stream', 'source', ret);
@@ -233,7 +255,7 @@ function pipeline(...streams) {
       }
     } else if (typeof stream === 'function') {
       ret = makeAsyncIterable(ret);
-      ret = stream(ret);
+      ret = stream(ret, signal);
 
       if (reading) {
         if (!isIterable(ret, true)) {
@@ -309,4 +331,4 @@ function pipeline(...streams) {
   return ret;
 }
 
-module.exports = pipeline;
+module.exports = { pipelineImpl, pipeline };
diff --git a/lib/stream.js b/lib/stream.js
@@ -29,7 +29,7 @@ const {
   promisify: { custom: customPromisify },
 } = require('internal/util');
 
-const pipeline = require('internal/streams/pipeline');
+const { pipeline } = require('internal/streams/pipeline');
 const { destroyer } = require('internal/streams/destroy');
 const eos = require('internal/streams/end-of-stream');
 const internalBuffer = require('internal/buffer');
diff --git a/lib/stream/promises.js b/lib/stream/promises.js
@@ -18,7 +18,7 @@ const {
   isStream,
 } = require('internal/streams/utils');
 
-const pl = require('internal/streams/pipeline');
+const { pipelineImpl: pl } = require('internal/streams/pipeline');
 const eos = require('internal/streams/end-of-stream');
 
 function pipeline(...streams) {
@@ -32,13 +32,13 @@ function pipeline(...streams) {
       validateAbortSignal(signal, 'options.signal');
     }
 
-    const pipe = pl(...streams, (err, value) => {
+    const pipe = pl(streams, (err, value) => {
       if (err) {
         reject(err);
       } else {
         resolve(value);
       }
-    });
+    }, { signal });
     if (signal) {
       addAbortSignalNoValidate(signal, pipe);
     }
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
@@ -11,10 +11,12 @@ const {
   Duplex,
   addAbortSignal,
 } = require('stream');
+const pipelinep = require('stream/promises').pipeline
 const assert = require('assert');
 const http = require('http');
 const { promisify } = require('util');
 const net = require('net');
+const tsp = require('timers/promises');
 
 {
   let finished = false;
@@ -1420,3 +1422,20 @@ const net = require('net');
 
   writableLike.emit('close');
 }
+
+{
+  const ac = new AbortController();
+  const signal = ac.signal;
+  pipelinep(
+    async function * (signal) {
+      await tsp.setTimeout(1e6, signal);
+    },
+    async function (source) {
+
+    },
+    { signal }
+  ).catch(common.mustCall((err) => {
+    assert.strictEqual(err.name, 'AbortError');
+  }));
+  ac.abort();
+}
