fixup: faster version

Author: ronag

lib/internal/streams/pipeline.js

@@ -146,37 +146,43 @@ async function* fromReadable(val) {
   yield* createReadableStreamAsyncIterator(val);
 }
 
+function pumpStdio(iterable, writable, finish) {
+  let iterator;
+  if (iterable[SymbolAsyncIterator])
+    iterator = iterable[SymbolAsyncIterator]();
+  else if (iterable[SymbolIterator])
+    iterator = iterable[SymbolIterator]();
+
+  async function _next(err) {
+    if (err) {
+      return finish(err);
+    }
+    try {
+      const { value, done } = await iterator.next();
+      if (done) {
+        return finish();
+      }
+      writable.write(value, _next);
+    } catch (err) {
+      finish(err);
+    }
+  }
+  _next();
+}
+
 async function pump(iterable, writable, finish) {
   if (!EE) {
     EE = require('events');
   }
   let error;
   try {
-    if (!isStdio(writable)) {
-      for await (const chunk of iterable) {
-        if (!writable.write(chunk)) {
-          if (writable.destroyed) return;
-          await EE.once(writable, 'drain');
-        }
-      }
-      writable.end();
-    } else {
-      const errorPromise = new Promise((resolve, reject) => {
-        writable.on('error', reject);
-      });
-      // Don't propagate to unhandledRejection
-      errorPromise.catch(() => {});
-      for await (const chunk of iterable) {
-        await Promise.race([
-          errorPromise,
-          new Promise((resolve, reject) => {
-            writable.write(chunk, null, (err) => {
-              err ? reject(err) : resolve();
-            });
-          })
-        ]);
+    for await (const chunk of iterable) {
+      if (!writable.write(chunk)) {
+        if (writable.destroyed) return;
+        await EE.once(writable, 'drain');
       }
     }
+    writable.end();
   } catch (err) {
     error = err;
   } finally {
@@ -292,7 +298,11 @@ function pipeline(...streams) {
         ret = makeAsyncIterable(ret);
 
         finishCount++;
-        pump(ret, stream, finish);
+        if (isStdio(stream)) {
+          pumpStdio(ret, stream, finish);
+        } else {
+          pump(ret, stream, finish);
+        }
       }
       ret = stream;
     } else {

test/parallel/test-stream-pipeline-process.js

@@ -21,7 +21,8 @@ if (process.argv[2] === 'child') {
     `"${process.execPath}"`,
     `"${__filename}"`,
     'child'
-  ].join(' '), common.mustCall((err) => {
+  ].join(' '), common.mustCall((err, stdout) => {
     assert.ifError(err);
+    assert.strictEqual(stdout, 'hello');
   }));
 }