stream: complete pipeline with stdio

stdio (stderr & stdout) should for compatibility
reasons not be closed/end():ed. However, this
causes pipeline with a stdio destination to
never finish. This commit fixes this issue at
a performance cost.

Refs: nodejs#7606

Fixes: nodejs#32363

lib/internal/streams/pipeline.js

@@ -6,7 +6,8 @@
 const {
   ArrayIsArray,
   SymbolAsyncIterator,
-  SymbolIterator
+  SymbolIterator,
+  Promise
 } = primordials;
 
 let eos;
@@ -147,13 +148,23 @@ async function pump(iterable, writable, finish) {
   }
   let error;
   try {
-    for await (const chunk of iterable) {
-      if (!writable.write(chunk)) {
-        if (writable.destroyed) return;
-        await EE.once(writable, 'drain');
+    if (writable !== process.stdout && writable !== process.stderr) {
+      for await (const chunk of iterable) {
+        if (!writable.write(chunk)) {
+          if (writable.destroyed) return;
+          await EE.once(writable, 'drain');
+        }
+      }
+      writable.end();
+    } else {
+      for await (const chunk of iterable) {
+        await new Promise((resolve, reject) => {
+          writable.write(chunk, null, (err) => {
+            err ? reject(err) : resolve();
+          });
+        });
       }
     }
-    writable.end();
   } catch (err) {
     error = err;
   } finally {
@@ -202,7 +213,9 @@ function pipeline(...streams) {
     const reading = i < streams.length - 1;
     const writing = i > 0;
 
-    if (isStream(stream)) {
+    if (stream === process.stdout || stream === process.stderr) {
+      // `pipe()` doesn't .end() stdout and stderr.
+    } else if (isStream(stream)) {
       finishCount++;
       destroys.push(destroyer(stream, reading, writing, !reading, finish));
     }
@@ -263,7 +276,10 @@ function pipeline(...streams) {
         destroys.push(destroyer(ret, false, true, true, finish));
       }
     } else if (isStream(stream)) {
-      if (isReadable(ret)) {
+      if (isReadable(ret) &&
+          stream !== process.stdout &&
+          stream !== process.stderr
+      ) {
         ret.pipe(stream);
       } else {
         ret = makeAsyncIterable(ret);