stream: duplexify

nodejs · Jul 26, 2021 · ce3b597 · ce3b597
1 parent ab03ab4
commit ce3b597
Show file tree

Hide file tree

Showing 7 changed files with 368 additions and 132 deletions.
diff --git a/lib/internal/streams/compose.js b/lib/internal/streams/compose.js
@@ -2,69 +2,38 @@
 
 const pipeline = require('internal/streams/pipeline');
 const Duplex = require('internal/streams/duplex');
-const { createDeferredPromise } = require('internal/util');
 const { destroyer } = require('internal/streams/destroy');
-const from = require('internal/streams/from');
 const {
   isNodeStream,
-  isIterable,
   isReadable,
   isWritable,
 } = require('internal/streams/utils');
-const {
-  PromiseResolve,
-} = primordials;
 const {
   AbortError,
   codes: {
-    ERR_INVALID_ARG_TYPE,
     ERR_INVALID_ARG_VALUE,
-    ERR_INVALID_RETURN_VALUE,
     ERR_MISSING_ARGS,
   },
 } = require('internal/errors');
-const assert = require('internal/assert');
-
-// This is needed for pre node 17.
-class ComposeDuplex extends Duplex {
-  constructor(options) {
-    super(options);
-
-    // https://github.com/nodejs/node/pull/34385
-
-    if (options?.readable === false) {
-      this._readableState.readable = false;
-      this._readableState.ended = true;
-      this._readableState.endEmitted = true;
-    }
-
-    if (options?.writable === false) {
-      this._writableState.writable = false;
-      this._writableState.ending = true;
-      this._writableState.ended = true;
-      this._writableState.finished = true;
-    }
-  }
-}
 
 module.exports = function compose(...streams) {
   if (streams.length === 0) {
     throw new ERR_MISSING_ARGS('streams');
   }
 
   if (streams.length === 1) {
-    return makeDuplex(streams[0], 'streams[0]');
+    return Duplex.from(streams[0]);
   }
 
   const orgStreams = [...streams];
 
   if (typeof streams[0] === 'function') {
-    streams[0] = makeDuplex(streams[0], 'streams[0]');
+    streams[0] = Duplex.from(streams[0]);
   }
 
   if (typeof streams[streams.length - 1] === 'function') {
     const idx = streams.length - 1;
-    streams[idx] = makeDuplex(streams[idx], `streams[${idx}]`);
+    streams[idx] = Duplex.from(streams[idx]);
   }
 
   for (let n = 0; n < streams.length; ++n) {
@@ -116,8 +85,8 @@ module.exports = function compose(...streams) {
   // TODO(ronag): Avoid double buffering.
   // Implement Writable/Readable/Duplex traits.
   // See, https://github.com/nodejs/node/pull/33515.
-  d = new ComposeDuplex({
-    highWaterMark: 1,
+  d = new Duplex({
+    // TODO (ronag): highWaterMark?
     writableObjectMode: !!head?.writableObjectMode,
     readableObjectMode: !!tail?.writableObjectMode,
     writable,
@@ -203,82 +172,3 @@ module.exports = function compose(...streams) {
 
   return d;
 };
-
-function makeDuplex(stream, name) {
-  let ret;
-  if (typeof stream === 'function') {
-    assert(stream.length > 0);
-
-    const { value, write, final } = fromAsyncGen(stream);
-
-    if (isIterable(value)) {
-      ret = from(ComposeDuplex, value, {
-        objectMode: true,
-        highWaterMark: 1,
-        write,
-        final
-      });
-    } else if (typeof value?.then === 'function') {
-      const promise = PromiseResolve(value)
-        .then((val) => {
-          if (val != null) {
-            throw new ERR_INVALID_RETURN_VALUE('nully', name, val);
-          }
-        })
-        .catch((err) => {
-          destroyer(ret, err);
-        });
-
-      ret = new ComposeDuplex({
-        objectMode: true,
-        highWaterMark: 1,
-        readable: false,
-        write,
-        final(cb) {
-          final(() => promise.then(cb, cb));
-        }
-      });
-    } else {
-      throw new ERR_INVALID_RETURN_VALUE(
-        'Iterable, AsyncIterable or AsyncFunction', name, value);
-    }
-  } else if (isNodeStream(stream)) {
-    ret = stream;
-  } else if (isIterable(stream)) {
-    ret = from(ComposeDuplex, stream, {
-      objectMode: true,
-      highWaterMark: 1,
-      writable: false
-    });
-  } else {
-    throw new ERR_INVALID_ARG_TYPE(
-      name,
-      ['Stream', 'Iterable', 'AsyncIterable', 'Function'],
-      stream)
-    ;
-  }
-  return ret;
-}
-
-function fromAsyncGen(fn) {
-  let { promise, resolve } = createDeferredPromise();
-  const value = fn(async function*() {
-    while (true) {
-      const { chunk, done, cb } = await promise;
-      process.nextTick(cb);
-      if (done) return;
-      yield chunk;
-      ({ promise, resolve } = createDeferredPromise());
-    }
-  }());
-
-  return {
-    value,
-    write(chunk, encoding, cb) {
-      resolve({ chunk, done: false, cb });
-    },
-    final(cb) {
-      resolve({ done: true, cb });
-    }
-  };
-}
diff --git a/lib/internal/streams/duplex.js b/lib/internal/streams/duplex.js
@@ -133,3 +133,12 @@ Duplex.fromWeb = function(pair, options) {
 Duplex.toWeb = function(duplex) {
   return lazyWebStreams().newReadableWritablePairFromDuplex(duplex);
 };
+
+let duplexify;
+
+Duplex.from = function(body) {
+  if (!duplexify) {
+    duplexify = require('internal/streams/duplexify');
+  }
+  return duplexify(body, 'body');
+};