stream: duplexify

PR-URL: https://github.com/nodejs/node/pull/39519
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>

doc/api/stream.md

@@ -2081,6 +2081,34 @@ added: REPLACEME
 * `streamWritable` {stream.Writable}
 * Returns: {WritableStream}
 
+### `stream.Duplex.from(src)`
+<!-- YAML
+added: REPLACEME
+-->
+
+* `src` {Stream|Blob|ArrayBuffer|string|Iterable|AsyncIterable|
+  AsyncGeneratorFunction|AsyncFunction|Promise|Object}
+
+A utility method for creating duplex streams.
+
+* `Stream` converts writable stream into writable `Duplex` and readable stream
+  to `Duplex`.
+* `Blob` converts into readable `Duplex`.
+* `string` converts into readable `Duplex`.
+* `ArrayBuffer` converts into readable `Duplex`.
+* `AsyncIterable` converts into a readable `Duplex`. Cannot yield
+  `null`.
+* `AsyncGeneratorFunction` converts into a readable/writable transform
+  `Duplex`. Must take a source `AsyncIterable` as first parameter. Cannot yield
+  `null`.
+* `AsyncFunction` converts into a writable `Duplex`. Must return
+  either `null` or `undefined`
+* `Object ({ writable, readable })` converts `readable` and
+  `writable` into `Stream` and then combines them into `Duplex` where the
+  `Duplex` will write to the `writable` and read from the `readable`.
+* `Promise` converts into readable `Duplex`. Value `null` is ignored.
+* Returns: {stream.Duplex}
+
 ### `stream.Duplex.fromWeb(pair[, options])`
 <!-- YAML
 added: REPLACEME

lib/internal/streams/compose.js

@@ -2,28 +2,19 @@
 
 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 {
@@ -53,18 +44,18 @@ module.exports = function compose(...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) {
@@ -117,7 +108,7 @@ module.exports = function compose(...streams) {
   // Implement Writable/Readable/Duplex traits.
   // See, https://github.com/nodejs/node/pull/33515.
   d = new ComposeDuplex({
-    highWaterMark: 1,
+    // TODO (ronag): highWaterMark?
     writableObjectMode: !!head?.writableObjectMode,
     readableObjectMode: !!tail?.writableObjectMode,
     writable,
@@ -203,82 +194,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 });
-    }
-  };
-}

lib/internal/streams/destroy.js

@@ -323,7 +323,7 @@ function emitErrorCloseLegacy(stream, err) {
 
 // Normalize destroy for legacy.
 function destroyer(stream, err) {
-  if (isDestroyed(stream)) {
+  if (!stream || isDestroyed(stream)) {
     return;
   }
 

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');
+};

lib/internal/streams/duplexify.js

@@ -0,0 +1,359 @@
+'use strict';
+
+const {
+  isReadable,
+  isWritable,
+  isIterable,
+  isNodeStream,
+  isReadableNodeStream,
+  isWritableNodeStream,
+  isDuplexNodeStream,
+} = require('internal/streams/utils');
+const eos = require('internal/streams/end-of-stream');
+const {
+  AbortError,
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+    ERR_INVALID_RETURN_VALUE,
+  },
+} = require('internal/errors');
+const { destroyer } = require('internal/streams/destroy');
+const Duplex = require('internal/streams/duplex');
+const Readable = require('internal/streams/readable');
+const { createDeferredPromise } = require('internal/util');
+const from = require('internal/streams/from');
+
+const {
+  isBlob,
+} = require('internal/blob');
+
+const {
+  FunctionPrototypeCall
+} = primordials;
+
+// This is needed for pre node 17.
+class Duplexify 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 duplexify(body, name) {
+  if (isDuplexNodeStream(body)) {
+    return body;
+  }
+
+  if (isReadableNodeStream(body)) {
+    return _duplexify({ readable: body });
+  }
+
+  if (isWritableNodeStream(body)) {
+    return _duplexify({ writable: body });
+  }
+
+  if (isNodeStream(body)) {
+    return _duplexify({ writable: false, readable: false });
+  }
+
+  // TODO: Webstreams
+  // if (isReadableStream(body)) {
+  //   return _duplexify({ readable: Readable.fromWeb(body) });
+  // }
+
+  // TODO: Webstreams
+  // if (isWritableStream(body)) {
+  //   return _duplexify({ writable: Writable.fromWeb(body) });
+  // }
+
+  if (typeof body === 'function') {
+    const { value, write, final } = fromAsyncGen(body);
+
+    if (isIterable(value)) {
+      return from(Duplexify, value, {
+        // TODO (ronag): highWaterMark?
+        objectMode: true,
+        write,
+        final
+      });
+    }
+
+    const then = value?.then;
+    if (typeof then === 'function') {
+      let d;
+
+      const promise = FunctionPrototypeCall(
+        then,
+        value,
+        (val) => {
+          if (val != null) {
+            throw new ERR_INVALID_RETURN_VALUE('nully', 'body', val);
+          }
+        },
+        (err) => {
+          destroyer(d, err);
+        }
+      );
+
+      return d = new Duplexify({
+        // TODO (ronag): highWaterMark?
+        objectMode: true,
+        readable: false,
+        write,
+        final(cb) {
+          final(async () => {
+            try {
+              await promise;
+              process.nextTick(cb, null);
+            } catch (err) {
+              process.nextTick(cb, err);
+            }
+          });
+        }
+      });
+    }
+
+    throw new ERR_INVALID_RETURN_VALUE(
+      'Iterable, AsyncIterable or AsyncFunction', name, value);
+  }
+
+  if (isBlob(body)) {
+    return duplexify(body.arrayBuffer());
+  }
+
+  if (isIterable(body)) {
+    return from(Duplexify, body, {
+      // TODO (ronag): highWaterMark?
+      objectMode: true,
+      writable: false
+    });
+  }
+
+  // TODO: Webstreams.
+  // if (
+  //   isReadableStream(body?.readable) &&
+  //   isWritableStream(body?.writable)
+  // ) {
+  //   return Duplexify.fromWeb(body);
+  // }
+
+  if (
+    typeof body?.writable === 'object' ||
+    typeof body?.readable === 'object'
+  ) {
+    const readable = body?.readable ?
+      isReadableNodeStream(body?.readable) ? body?.readable :
+        duplexify(body.readable) :
+      undefined;
+
+    const writable = body?.writable ?
+      isWritableNodeStream(body?.writable) ? body?.writable :
+        duplexify(body.writable) :
+      undefined;
+
+    return _duplexify({ readable, writable });
+  }
+
+  const then = body?.then;
+  if (typeof then === 'function') {
+    let d;
+
+    FunctionPrototypeCall(
+      then,
+      body,
+      (val) => {
+        if (val != null) {
+          d.push(val);
+        }
+        d.push(null);
+      },
+      (err) => {
+        destroyer(d, err);
+      }
+    );
+
+    return d = new Duplexify({
+      objectMode: true,
+      writable: false,
+      read() {}
+    });
+  }
+
+  throw new ERR_INVALID_ARG_TYPE(
+    name,
+    ['Blob', 'ReadableStream', 'WritableStream', 'Stream', 'Iterable',
+     'AsyncIterable', 'Function', '{ readable, writable } pair', 'Promise'],
+    body);
+};
+
+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 });
+    }
+  };
+}
+
+function _duplexify(pair) {
+  const r = pair.readable && typeof pair.readable.read !== 'function' ?
+    Readable.wrap(pair.readable) : pair.readable;
+  const w = pair.writable;
+
+  let readable = !!isReadable(r);
+  let writable = !!isWritable(w);
+
+  let ondrain;
+  let onfinish;
+  let onreadable;
+  let onclose;
+  let d;
+
+  function onfinished(err) {
+    const cb = onclose;
+    onclose = null;
+
+    if (cb) {
+      cb(err);
+    } else if (err) {
+      d.destroy(err);
+    } else if (!readable && !writable) {
+      d.destroy();
+    }
+  }
+
+  // TODO(ronag): Avoid double buffering.
+  // Implement Writable/Readable/Duplex traits.
+  // See, https://github.com/nodejs/node/pull/33515.
+  d = new Duplexify({
+    // TODO (ronag): highWaterMark?
+    readableObjectMode: !!r?.readableObjectMode,
+    writableObjectMode: !!w?.writableObjectMode,
+    readable,
+    writable,
+  });
+
+  if (writable) {
+    eos(w, (err) => {
+      writable = false;
+      if (err) {
+        destroyer(r, err);
+      }
+      onfinished(err);
+    });
+
+    d._write = function(chunk, encoding, callback) {
+      if (w.write(chunk, encoding)) {
+        callback();
+      } else {
+        ondrain = callback;
+      }
+    };
+
+    d._final = function(callback) {
+      w.end();
+      onfinish = callback;
+    };
+
+    w.on('drain', function() {
+      if (ondrain) {
+        const cb = ondrain;
+        ondrain = null;
+        cb();
+      }
+    });
+
+    w.on('finish', function() {
+      if (onfinish) {
+        const cb = onfinish;
+        onfinish = null;
+        cb();
+      }
+    });
+  }
+
+  if (readable) {
+    eos(r, (err) => {
+      readable = false;
+      if (err) {
+        destroyer(r, err);
+      }
+      onfinished(err);
+    });
+
+    r.on('readable', function() {
+      if (onreadable) {
+        const cb = onreadable;
+        onreadable = null;
+        cb();
+      }
+    });
+
+    r.on('end', function() {
+      d.push(null);
+    });
+
+    d._read = function() {
+      while (true) {
+        const buf = r.read();
+
+        if (buf === null) {
+          onreadable = d._read;
+          return;
+        }
+
+        if (!d.push(buf)) {
+          return;
+        }
+      }
+    };
+  }
+
+  d._destroy = function(err, callback) {
+    if (!err && onclose !== null) {
+      err = new AbortError();
+    }
+
+    onreadable = null;
+    ondrain = null;
+    onfinish = null;
+
+    if (onclose === null) {
+      callback(err);
+    } else {
+      onclose = callback;
+      destroyer(w, err);
+      destroyer(r, err);
+    }
+  };
+
+  return d;
+}

lib/internal/streams/end-of-stream.js

@@ -25,6 +25,7 @@ const {
   isWritable,
   isWritableNodeStream,
   isWritableFinished,
+  isNodeStream,
   willEmitClose: _willEmitClose,
 } = require('internal/streams/utils');
 
@@ -53,6 +54,13 @@ function eos(stream, options, callback) {
   const writable = options.writable ||
     (options.writable !== false && isWritableNodeStream(stream));
 
+  if (isNodeStream(stream)) {
+    // Do nothing...
+  } else {
+    // TODO: Webstreams.
+    // TODO: Throw INVALID_ARG_TYPE.
+  }
+
   const wState = stream._writableState;
   const rState = stream._readableState;
 

lib/internal/streams/pipeline.js

@@ -10,9 +10,9 @@ const {
 } = primordials;
 
 const eos = require('internal/streams/end-of-stream');
-
 const { once } = require('internal/util');
 const destroyImpl = require('internal/streams/destroy');
+const Duplex = require('internal/streams/duplex');
 const {
   aggregateTwoErrors,
   codes: {
@@ -219,9 +219,7 @@ function pipeline(...streams) {
       } else if (isIterable(stream) || isReadableNodeStream(stream)) {
         ret = stream;
       } else {
-        throw new ERR_INVALID_ARG_TYPE(
-          'source', ['Stream', 'Iterable', 'AsyncIterable', 'Function'],
-          stream);
+        ret = Duplex.from(stream);
       }
     } else if (typeof stream === 'function') {
       ret = makeAsyncIterable(ret);
@@ -289,9 +287,7 @@ function pipeline(...streams) {
       }
       ret = stream;
     } else {
-      const name = reading ? `transform[${i - 1}]` : 'destination';
-      throw new ERR_INVALID_ARG_TYPE(
-        name, ['Stream', 'Function'], stream);
+      ret = Duplex.from(stream);
     }
   }
 

lib/internal/streams/readable.js

@@ -1097,15 +1097,7 @@ Readable.prototype.iterator = function(options) {
 
 function streamToAsyncIterator(stream, options) {
   if (typeof stream.read !== 'function') {
-    // v1 stream
-    const src = stream;
-    stream = new Readable({
-      objectMode: true,
-      destroy(err, callback) {
-        destroyImpl.destroyer(src, err);
-        callback(err);
-      }
-    }).wrap(src);
+    stream = Readable.wrap(stream, { objectMode: true });
   }
 
   const iter = createAsyncIterator(stream, options);
@@ -1391,3 +1383,14 @@ Readable.fromWeb = function(readableStream, options) {
 Readable.toWeb = function(streamReadable) {
   return lazyWebStreams().newReadableStreamFromStreamReadable(streamReadable);
 };
+
+Readable.wrap = function(src, options) {
+  return new Readable({
+    objectMode: src.readableObjectMode ?? src.objectMode ?? true,
+    ...options,
+    destroy(err, callback) {
+      destroyImpl.destroyer(src, err);
+      callback(err);
+    }
+  }).wrap(src);
+};

lib/internal/streams/utils.js

@@ -27,6 +27,15 @@ function isWritableNodeStream(obj) {
   );
 }
 
+function isDuplexNodeStream(obj) {
+  return !!(
+    obj &&
+    (typeof obj.pipe === 'function' && obj._readableState) &&
+    typeof obj.on === 'function' &&
+    typeof obj.write === 'function'
+  );
+}
+
 function isNodeStream(obj) {
   return (
     obj &&
@@ -102,14 +111,14 @@ function isReadableFinished(stream, strict) {
 
 function isReadable(stream) {
   const r = isReadableNodeStream(stream);
-  if (r === null || typeof stream.readable !== 'boolean') return null;
+  if (r === null || typeof stream?.readable !== 'boolean') return null;
   if (isDestroyed(stream)) return false;
   return r && stream.readable && !isReadableFinished(stream);
 }
 
 function isWritable(stream) {
   const r = isWritableNodeStream(stream);
-  if (r === null || typeof stream.writable !== 'boolean') return null;
+  if (r === null || typeof stream?.writable !== 'boolean') return null;
   if (isDestroyed(stream)) return false;
   return r && stream.writable && !isWritableEnded(stream);
 }
@@ -199,6 +208,7 @@ module.exports = {
   kDestroyed,
   isClosed,
   isDestroyed,
+  isDuplexNodeStream,
   isFinished,
   isIterable,
   isReadable,

test/parallel/test-stream-duplex-from.js

@@ -0,0 +1,120 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { Duplex, Readable, Writable } = require('stream');
+
+{
+  const d = Duplex.from({
+    readable: new Readable({
+      read() {
+        this.push('asd');
+        this.push(null);
+      }
+    })
+  });
+  assert.strictEqual(d.readable, true);
+  assert.strictEqual(d.writable, false);
+  d.once('readable', common.mustCall(function() {
+    assert.strictEqual(d.read().toString(), 'asd');
+  }));
+  d.once('end', common.mustCall(function() {
+    assert.strictEqual(d.readable, false);
+  }));
+}
+
+{
+  const d = Duplex.from(new Readable({
+    read() {
+      this.push('asd');
+      this.push(null);
+    }
+  }));
+  assert.strictEqual(d.readable, true);
+  assert.strictEqual(d.writable, false);
+  d.once('readable', common.mustCall(function() {
+    assert.strictEqual(d.read().toString(), 'asd');
+  }));
+  d.once('end', common.mustCall(function() {
+    assert.strictEqual(d.readable, false);
+  }));
+}
+
+{
+  let ret = '';
+  const d = Duplex.from(new Writable({
+    write(chunk, encoding, callback) {
+      ret += chunk;
+      callback();
+    }
+  }));
+  assert.strictEqual(d.readable, false);
+  assert.strictEqual(d.writable, true);
+  d.end('asd');
+  d.on('finish', common.mustCall(function() {
+    assert.strictEqual(d.writable, false);
+    assert.strictEqual(ret, 'asd');
+  }));
+}
+
+{
+  let ret = '';
+  const d = Duplex.from({
+    writable: new Writable({
+      write(chunk, encoding, callback) {
+        ret += chunk;
+        callback();
+      }
+    })
+  });
+  assert.strictEqual(d.readable, false);
+  assert.strictEqual(d.writable, true);
+  d.end('asd');
+  d.on('finish', common.mustCall(function() {
+    assert.strictEqual(d.writable, false);
+    assert.strictEqual(ret, 'asd');
+  }));
+}
+
+{
+  let ret = '';
+  const d = Duplex.from({
+    readable: new Readable({
+      read() {
+        this.push('asd');
+        this.push(null);
+      }
+    }),
+    writable: new Writable({
+      write(chunk, encoding, callback) {
+        ret += chunk;
+        callback();
+      }
+    })
+  });
+  assert.strictEqual(d.readable, true);
+  assert.strictEqual(d.writable, true);
+  d.once('readable', common.mustCall(function() {
+    assert.strictEqual(d.read().toString(), 'asd');
+  }));
+  d.once('end', common.mustCall(function() {
+    assert.strictEqual(d.readable, false);
+  }));
+  d.end('asd');
+  d.once('finish', common.mustCall(function() {
+    assert.strictEqual(d.writable, false);
+    assert.strictEqual(ret, 'asd');
+  }));
+}
+
+{
+  const d = Duplex.from(Promise.resolve('asd'));
+  assert.strictEqual(d.readable, true);
+  assert.strictEqual(d.writable, false);
+  d.once('readable', common.mustCall(function() {
+    assert.strictEqual(d.read().toString(), 'asd');
+  }));
+  d.once('end', common.mustCall(function() {
+    assert.strictEqual(d.readable, false);
+  }));
+}

tools/doc/type-parser.mjs

@@ -34,6 +34,10 @@ const customTypesMap = {
 
   'AsyncIterable': 'https://tc39.github.io/ecma262/#sec-asynciterable-interface',
 
+  'AsyncFunction': 'https://tc39.es/ecma262/#sec-async-function-constructor',
+
+  'AsyncGeneratorFunction': 'https://tc39.es/proposal-async-iteration/#sec-asyncgeneratorfunction-constructor',
+
   'bigint': `${jsDocPrefix}Reference/Global_Objects/BigInt`,
   'WebAssembly.Instance':
     `${jsDocPrefix}Reference/Global_Objects/WebAssembly/Instance`,