fixup: end stdio

Author: ronag

lib/internal/streams/pipeline.js

@@ -6,8 +6,7 @@
 const {
   ArrayIsArray,
   SymbolAsyncIterator,
-  SymbolIterator,
-  Promise
+  SymbolIterator
 } = primordials;
 
 let eos;
@@ -114,10 +113,6 @@ function isStream(obj) {
   return isReadable(obj) || isWritable(obj);
 }
 
-function isStdio(obj) {
-  return obj === process.stdout || obj === process.stderr;
-}
-
 function isIterable(obj, isAsync) {
   if (!obj) return false;
   if (isAsync === true) return typeof obj[SymbolAsyncIterator] === 'function';
@@ -151,32 +146,14 @@ async function pump(iterable, writable, finish) {
     EE = require('events');
   }
   let error;
-  writable.on('error', (err) => {
-    error = err;
-  });
   try {
-    let prev;
-    for await (const next of iterable) {
-      if (prev != null) {
-        if (!writable.write(prev)) {
-          if (writable.destroyed) return;
-          await EE.once(writable, 'drain');
-        }
+    for await (const chunk of iterable) {
+      if (!writable.write(chunk)) {
+        if (writable.destroyed) return;
+        await EE.once(writable, 'drain');
       }
-      prev = next;
-    }
-
-    if (prev != null) {
-      await new Promise((resolve, reject) => {
-        writable.write(prev, (err) => {
-          err ? reject(err) : resolve();
-        });
-      });
-    }
-
-    if (!isStdio(writable)) {
-      writable.end();
     }
+    writable.end();
   } catch (err) {
     error = err;
   } finally {
@@ -225,7 +202,7 @@ function pipeline(...streams) {
     const reading = i < streams.length - 1;
     const writing = i > 0;
 
-    if (isStream(stream) && !isStdio(stream)) {
+    if (isStream(stream)) {
       finishCount++;
       destroys.push(destroyer(stream, reading, writing, !reading, finish));
     }
@@ -286,8 +263,15 @@ function pipeline(...streams) {
         destroys.push(destroyer(ret, false, true, true, finish));
       }
     } else if (isStream(stream)) {
-      if (isReadable(ret) && !isStdio(stream)) {
+      if (isReadable(ret)) {
         ret.pipe(stream);
+
+        // Compat. Before node v10.12.0 stdio used to throw an error so
+        // pipe() did/does not end() stdio destinations.
+        // Now they allow it but "secretly" don't close the underlying fd.
+        if (stream === process.stdout || stream === process.stderr) {
+          ret.on('end', () => stream.end());
+        }
       } else {
         ret = makeAsyncIterable(ret);