stream: support array of streams in promises pipeline

Fixes: #40191

PR-URL: #40193
Reviewed-By: Luigi Pinca <[email protected]>
Reviewed-By: Robert Nagy <[email protected]>
Reviewed-By: James M Snell <[email protected]>

lib/internal/streams/pipeline.js

@@ -166,17 +166,14 @@ async function pump(iterable, writable, finish) {
 }
 
 function pipeline(...streams) {
-  const callback = once(popCallback(streams));
+  return pipelineImpl(streams, once(popCallback(streams)));
+}
 
-  // stream.pipeline(streams, callback)
-  if (ArrayIsArray(streams[0]) && streams.length === 1) {
+function pipelineImpl(streams, callback, opts) {
+  if (streams.length === 1 && ArrayIsArray(streams[0])) {
     streams = streams[0];
   }
 
-  return pipelineImpl(streams, callback);
-}
-
-function pipelineImpl(streams, callback, opts) {
   if (streams.length < 2) {
     throw new ERR_MISSING_ARGS('streams');
   }

test/parallel/test-stream-pipeline.js

@@ -1406,3 +1406,44 @@ const tsp = require('timers/promises');
   }));
   ac.abort();
 }
+
+{
+  async function run() {
+    let finished = false;
+    let text = '';
+    const write = new Writable({
+      write(data, enc, cb) {
+        text += data;
+        cb();
+      }
+    });
+    write.on('finish', () => {
+      finished = true;
+    });
+
+    await pipelinep([Readable.from('Hello World!'), write]);
+    assert(finished);
+    assert.strictEqual(text, 'Hello World!');
+  }
+
+  run();
+}
+
+{
+  let finished = false;
+  let text = '';
+  const write = new Writable({
+    write(data, enc, cb) {
+      text += data;
+      cb();
+    }
+  });
+  write.on('finish', () => {
+    finished = true;
+  });
+
+  pipeline([Readable.from('Hello World!'), write], common.mustSucceed(() => {
+    assert(finished);
+    assert.strictEqual(text, 'Hello World!');
+  }));
+}