From ea09812160e5fcead3e6617adb2a9d2b6b86db42 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sun, 25 Jul 2021 20:01:43 +0200
Subject: [PATCH] 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 <ckbenjamingr@gmail.com>
Backport-PR-URL: https://github.com/nodejs/node/pull/39820
---
 doc/api/stream.md                        |  28 ++
 lib/internal/streams/destroy.js          |   1 +
 lib/internal/streams/duplex.js           |   9 +
 lib/internal/streams/duplexify.js        | 372 +++++++++++++++++++++++
 lib/internal/streams/pipeline.js         |  10 +-
 lib/internal/streams/readable.js         |  21 +-
 lib/internal/streams/utils.js            |  53 ++++
 test/parallel/test-stream-duplex-from.js | 120 ++++++++
 tools/doc/type-parser.mjs                |   4 +
 9 files changed, 602 insertions(+), 16 deletions(-)
 create mode 100644 lib/internal/streams/duplexify.js
 create mode 100644 test/parallel/test-stream-duplex-from.js

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 79f524c152f3ce..26734fad48b40e 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -1966,6 +1966,34 @@ added: REPLACEME
 
 Returns whether the stream has been read from or cancelled.
 
+### `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.addAbortSignal(signal, stream)`
 <!-- YAML
 added: v15.4.0
diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
index 6d50b09f0948b4..06c74db830ed02 100644
--- a/lib/internal/streams/destroy.js
+++ b/lib/internal/streams/destroy.js
@@ -307,6 +307,7 @@ function isRequest(stream) {
 
 // Normalize destroy for legacy.
 function destroyer(stream, err) {
+  if (!stream) return;
   if (isRequest(stream)) return stream.abort();
   if (isRequest(stream.req)) return stream.req.abort();
   if (typeof stream.destroy === 'function') return stream.destroy(err);
diff --git a/lib/internal/streams/duplex.js b/lib/internal/streams/duplex.js
index 596d30a90a0418..c7b3056de4a8ec 100644
--- a/lib/internal/streams/duplex.js
+++ b/lib/internal/streams/duplex.js
@@ -108,3 +108,12 @@ ObjectDefineProperties(Duplex.prototype, {
     }
   }
 });
+
+let duplexify;
+
+Duplex.from = function(body) {
+  if (!duplexify) {
+    duplexify = require('internal/streams/duplexify');
+  }
+  return duplexify(body, 'body');
+};
diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js
new file mode 100644
index 00000000000000..f75393244a6495
--- /dev/null
+++ b/lib/internal/streams/duplexify.js
@@ -0,0 +1,372 @@
+'use strict';
+
+const {
+  isIterable,
+  isNodeStream,
+  isDestroyed,
+  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;
+
+function isReadable(stream) {
+  const r = isReadableNodeStream(stream);
+  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 (isDestroyed(stream)) return false;
+  return r && stream.writable && !isWritableEnded(stream);
+}
+
+// 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;
+}
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
index 5759dbd4a580a3..5cd1399cddcf0f 100644
--- a/lib/internal/streams/pipeline.js
+++ b/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) || isReadable(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);
     }
   }
 
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index 0e73c4320e5f4d..03953c37eac280 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -1077,15 +1077,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);
@@ -1381,3 +1373,14 @@ function endWritableNT(stream) {
 Readable.from = function(iterable, opts) {
   return from(Readable, iterable, opts);
 };
+
+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);
+};
diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js
index 96fd2d4e3580fc..a220dd43f53178 100644
--- a/lib/internal/streams/utils.js
+++ b/lib/internal/streams/utils.js
@@ -22,6 +22,54 @@ function isStream(obj) {
   return isReadable(obj) || isWritable(obj);
 }
 
+function isDuplexNodeStream(obj) {
+  return !!(
+    obj &&
+    (typeof obj.pipe === 'function' && obj._readableState) &&
+    typeof obj.on === 'function' &&
+    typeof obj.write === 'function'
+  );
+}
+
+function isReadableNodeStream(obj) {
+  return !!(
+    obj &&
+    typeof obj.pipe === 'function' &&
+    typeof obj.on === 'function' &&
+    (!obj._writableState || obj._readableState?.readable !== false) && // Duplex
+    (!obj._writableState || obj._readableState) // Writable has .pipe.
+  );
+}
+
+function isWritableNodeStream(obj) {
+  return !!(
+    obj &&
+    typeof obj.write === 'function' &&
+    typeof obj.on === 'function' &&
+    (!obj._readableState || obj._writableState?.writable !== false) // Duplex
+  );
+}
+
+function isNodeStream(obj) {
+  return (
+    obj &&
+    (
+      obj._readableState ||
+      obj._writableState ||
+      (typeof obj.write === 'function' && typeof obj.on === 'function') ||
+      (typeof obj.pipe === 'function' && typeof obj.on === 'function')
+    )
+  );
+}
+
+function isDestroyed(stream) {
+  if (!isNodeStream(stream)) return null;
+  const wState = stream._writableState;
+  const rState = stream._readableState;
+  const state = wState || rState;
+  return !!(stream.destroyed || state?.destroyed);
+}
+
 function isIterable(obj, isAsync) {
   if (obj == null) return false;
   if (isAsync === true) return typeof obj[SymbolAsyncIterator] === 'function';
@@ -41,6 +89,11 @@ function isDisturbed(stream) {
 module.exports = {
   isDisturbed,
   kIsDisturbed,
+  isDestroyed,
+  isReadableNodeStream,
+  isWritableNodeStream,
+  isDuplexNodeStream,
+  isNodeStream,
   isIterable,
   isReadable,
   isStream,
diff --git a/test/parallel/test-stream-duplex-from.js b/test/parallel/test-stream-duplex-from.js
new file mode 100644
index 00000000000000..265b61dfd062f9
--- /dev/null
+++ b/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);
+  }));
+}
diff --git a/tools/doc/type-parser.mjs b/tools/doc/type-parser.mjs
index 481a35ca330c80..fb396070266831 100644
--- a/tools/doc/type-parser.mjs
+++ b/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`,