stream: simplify pipeline

ronag · ronag · commit c6bf9539cc16 · 2020-01-26T17:53:10.000+01:00
PR-URL: nodejs#31316 Reviewed-By: Anna Henningsen <anna@addaleax.net> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Rich Trott <rtrott@gmail.com> Reviewed-By: Minwoo Jung <nodecorelab@gmail.com>
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
@@ -28,14 +28,6 @@ function isRequest(stream) {
   return stream && stream.setHeader && typeof stream.abort === 'function';
 }
 
-function destroyStream(stream, err) {
-  // request.destroy just do .end - .abort is what we want
-  if (isRequest(stream)) return stream.abort();
-  if (isRequest(stream.req)) return stream.req.abort();
-  if (typeof stream.destroy === 'function') return stream.destroy(err);
-  if (typeof stream.close === 'function') return stream.close();
-}
-
 function destroyer(stream, reading, writing, callback) {
   callback = once(callback);
 
@@ -57,7 +49,11 @@ function destroyer(stream, reading, writing, callback) {
     if (destroyed) return;
     destroyed = true;
 
-    destroyStream(stream, err);
+    // request.destroy just do .end - .abort is what we want
+    if (isRequest(stream)) return stream.abort();
+    if (isRequest(stream.req)) return stream.req.abort();
+    if (typeof stream.destroy === 'function') return stream.destroy(err);
+    if (typeof stream.close === 'function') return stream.close();
 
     callback(err || new ERR_STREAM_DESTROYED('pipe'));
   };
@@ -101,39 +97,19 @@ function makeAsyncIterable(val) {
     return val;
   } else if (isReadable(val)) {
     // Legacy streams are not Iterable.
-    return _fromReadable(val);
+    return fromReadable(val);
   } else {
     throw new ERR_INVALID_ARG_TYPE(
       'val', ['Readable', 'Iterable', 'AsyncIterable'], val);
   }
 }
 
-async function* _fromReadable(val) {
+async function* fromReadable(val) {
   if (!createReadableStreamAsyncIterator) {
     createReadableStreamAsyncIterator =
       require('internal/streams/async_iterator');
   }
-
-  try {
-    if (typeof val.read !== 'function') {
-      // createReadableStreamAsyncIterator does not support
-      // v1 streams. Convert it into a v2 stream.
-
-      if (!PassThrough) {
-        PassThrough = require('_stream_passthrough');
-      }
-
-      const pt = new PassThrough();
-      val
-        .on('error', (err) => pt.destroy(err))
-        .pipe(pt);
-      yield* createReadableStreamAsyncIterator(pt);
-    } else {
-      yield* createReadableStreamAsyncIterator(val);
-    }
-  } finally {
-    destroyStream(val);
-  }
+  yield* createReadableStreamAsyncIterator(val);
 }
 
 async function pump(iterable, writable, finish) {
