From 54d0ed7a732c7514bbfe27bb40df6ed8dee31cb1 Mon Sep 17 00:00:00 2001
From: Nitzan Uziely <nitzan@testim.io>
Date: Sat, 13 Feb 2021 17:15:23 +0200
Subject: [PATCH] stream: add AbortSignal to promisified pipeline

add support for AbortSignal to promisified pipeline.

Resolves: https://github.com/nodejs/node/issues/37321
---
 lib/stream/promises.js                | 22 ++++++++-
 test/parallel/test-stream-pipeline.js | 71 +++++++++++++++++++++++++++
 2 files changed, 92 insertions(+), 1 deletion(-)

diff --git a/lib/stream/promises.js b/lib/stream/promises.js
index 986db2e1f8db8a..26be4408638fc8 100644
--- a/lib/stream/promises.js
+++ b/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);
+    }
   });
 }
 
diff --git a/test/parallel/test-stream-pipeline.js b/test/parallel/test-stream-pipeline.js
index 78057f9eeffec6..3c3fdab2566983 100644
--- a/test/parallel/test-stream-pipeline.js
+++ b/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() {}