stream: add AbortSignal to promisified pipeline

add support for AbortSignal to promisified pipeline.

Resolves: #37321

Author: Nitzan Uziely

lib/stream/promises.js

@@ -1,22 +1,42 @@
 'use strict';
 
 const {
+  ArrayPrototypePop,
   Promise,
 } = primordials;
 
+const {
+  addAbortSignalNoValidate,
+} = require('internal/streams/add-abort-signal');
+
+const {
+  validateAbortSignal,
+} = require('internal/validators');
+
 let pl;
 let eos;
 
 function pipeline(...streams) {
   if (!pl) pl = require('internal/streams/pipeline');
+  let signal;
+  const lastArg = streams[streams.length - 1];
+  if (typeof lastArg === 'object' && 'aborted' in lastArg) {
+    signal = ArrayPrototypePop(streams);
+  }
   return new Promise((resolve, reject) => {
-    pl(...streams, (err, value) => {
+    if (signal) {
+      validateAbortSignal(signal);
+    }
+    const pipe = pl(...streams, (err, value) => {
       if (err) {
         reject(err);
       } else {
         resolve(value);
       }
     });
+    if (signal) {
+      addAbortSignalNoValidate(signal, pipe);
+    }
   });
 }
 

test/parallel/test-stream-pipeline.js

@@ -469,6 +469,77 @@ const net = require('net');
   run();
 }
 
+{
+  // Check aborted signal without values
+  const pipelinePromise = promisify(pipeline);
+  async function run() {
+    const ac = new AbortController();
+    const { signal } = ac;
+    async function* producer() {
+      ac.abort();
+      await Promise.resolve();
+      yield '8';
+    }
+
+    const w = new Writable({
+      write(chunk, encoding, callback) {
+        callback();
+      }
+    });
+    await pipelinePromise(producer, w, signal);
+  }
+
+  assert.rejects(run, { name: 'AbortError' }).then(common.mustCall());
+}
+
+{
+  // Check aborted signal after init.
+  const pipelinePromise = promisify(pipeline);
+  async function run() {
+    const ac = new AbortController();
+    const { signal } = ac;
+    async function* producer() {
+      yield '5';
+      await Promise.resolve();
+      ac.abort();
+      await Promise.resolve();
+      yield '8';
+    }
+
+    const w = new Writable({
+      write(chunk, encoding, callback) {
+        callback();
+      }
+    });
+    await pipelinePromise(producer, w, signal);
+  }
+
+  assert.rejects(run, { name: 'AbortError' }).then(common.mustCall());
+}
+
+{
+  // Check pre-aborted signal
+  const pipelinePromise = promisify(pipeline);
+  async function run() {
+    const signal = new EventTarget();
+    signal.aborted = true;
+    async function* producer() {
+      yield '5';
+      await Promise.resolve();
+      yield '8';
+    }
+
+    const w = new Writable({
+      write(chunk, encoding, callback) {
+        callback();
+      }
+    });
+    await pipelinePromise(producer, w, signal);
+  }
+
+  assert.rejects(run, { name: 'AbortError' }).then(common.mustCall());
+}
+
 {
   const read = new Readable({
     read() {}