From 96d11f348bb5b895dd3a2178e5c1cfa60d29048b Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sun, 25 Jul 2021 20:01:43 +0200
Subject: [PATCH 01/18] stream: duplexify

---
 lib/internal/streams/compose.js       | 120 +--------
 lib/internal/streams/duplex.js        |   9 +
 lib/internal/streams/duplexify.js     | 335 ++++++++++++++++++++++++++
 lib/internal/streams/end-of-stream.js |  13 +
 lib/internal/streams/pipeline.js      |  10 +-
 lib/internal/streams/readable.js      |  21 +-
 lib/internal/streams/utils.js         |  10 +
 7 files changed, 387 insertions(+), 131 deletions(-)
 create mode 100644 lib/internal/streams/duplexify.js

diff --git a/lib/internal/streams/compose.js b/lib/internal/streams/compose.js
index bdfdc4cebe0c47..65a36e17710935 100644
--- a/lib/internal/streams/compose.js
+++ b/lib/internal/streams/compose.js
@@ -2,50 +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 {
-  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) {
@@ -53,18 +22,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) {
@@ -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
index 381628adba3274..bc92e9021ab57c 100644
--- 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');
+};
diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js
new file mode 100644
index 00000000000000..7c71824bf76844
--- /dev/null
+++ b/lib/internal/streams/duplexify.js
@@ -0,0 +1,335 @@
+'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 Writable = require('internal/streams/writable');
+const { createDeferredPromise } = require('internal/util');
+const from = require('internal/streams/from');
+
+const {
+  isBlob,
+} = require('internal/blob');
+
+const {
+  isBrandCheck,
+} = require('internal/webstreams/util');
+
+const isReadableStream =
+  isBrandCheck('ReadableStream');
+const isWritableStream =
+  isBrandCheck('WritableStream');
+
+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;
+  } else if (isReadableNodeStream(body)) {
+    return _duplexify({ readable: body });
+  } else if (isWritableNodeStream(body)) {
+    return _duplexify({ writable: body });
+  } else if (isNodeStream(body)) {
+    return _duplexify({ writable: false, readable: false });
+  } else if (isReadableStream(body)) {
+    return _duplexify({ readable: Readable.fromWeb(body) });
+  } else if (isWritableStream(body)) {
+    return _duplexify({ writable: Writable.fromWeb(body) });
+  } else 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;
+    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(() => promise.then(cb, cb));
+        }
+      });
+    }
+
+    throw new ERR_INVALID_RETURN_VALUE(
+      'Iterable, AsyncIterable or AsyncFunction', name, value);
+  } else if (isBlob(body)) {
+    return duplexify(body.arrayBuffer());
+  } else if (isIterable(body)) {
+    return from(Duplexify, body, {
+      // TODO (ronag): highWaterMark?
+      objectMode: true,
+      writable: false
+    });
+  } else if (
+    isReadableStream(body?.readable) &&
+    isWritableStream(body?.writable)
+  ) {
+    return Duplexify.fromWeb(body);
+  } else if (
+    typeof body?.writable === 'object' ||
+    typeof body?.readable === 'object'
+  ) {
+    const readable = body?.readable ?
+      isReadableNodeStream(body?.readable) ? body?.readable :
+        duplexify(body.readable, body.options ?? body.readableOptions) :
+      undefined;
+
+    const writable = body?.writable ?
+      isWritableNodeStream(body?.writable) ? body?.writable :
+        duplexify(body.writable, body.options ?? body.writableOptions) :
+      undefined;
+
+    return _duplexify({ readable, writable });
+  } else if (typeof body?.then === 'function') {
+    let d;
+
+    FunctionPrototypeCall(
+      body.then,
+      body,
+      (val) => {
+        if (val != null) {
+          d.push(val);
+        }
+        d.push(null);
+      },
+      (err) => {
+        destroyer(d, err);
+      }
+    );
+
+    return d = new Duplexify({
+      objectMode: true,
+      writable: false
+    });
+  }
+
+  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 = 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();
+    }
+  }
+
+  eos(r, (err) => {
+    readable = false;
+    if (err) {
+      destroyer(w, err);
+    }
+    onfinished(err);
+  });
+
+  eos(w, (err) => {
+    writable = false;
+    if (err) {
+      destroyer(r, err);
+    }
+    onfinished(err);
+  });
+
+  // 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) {
+    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) {
+    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/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index 274c2796edd443..3c1f0b27971927 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -19,15 +19,20 @@ const {
 
 const {
   isClosed,
+  isReadableStream,
+  isWritableStream,
   isReadable,
   isReadableNodeStream,
   isReadableFinished,
   isWritable,
   isWritableNodeStream,
   isWritableFinished,
+  isNodeStream,
   willEmitClose: _willEmitClose,
 } = require('internal/streams/utils');
 
+const Duplex = require('./duplex');
+
 function isRequest(stream) {
   return stream.setHeader && typeof stream.abort === 'function';
 }
@@ -53,6 +58,14 @@ function eos(stream, options, callback) {
   const writable = options.writable ||
     (options.writable !== false && isWritableNodeStream(stream));
 
+  if (isNodeStream(stream)) {
+    // Do nothing...
+  } if (isReadableStream(stream) || isWritableStream(stream)) {
+    stream = Duplex.from(stream);
+  } else {
+    // TODO: Throw INVALID_ARG_TYPE.
+  }
+
   const wState = stream._writableState;
   const rState = stream._readableState;
 
diff --git a/lib/internal/streams/pipeline.js b/lib/internal/streams/pipeline.js
index 1b56c08b9e6958..3d39b3ac7b228a 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) || 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);
     }
   }
 
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index e7dd60a6d78c13..2d2403b63c7b5f 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -1099,15 +1099,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);
@@ -1388,3 +1380,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);
+};
diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js
index 01396b5113340f..e6543c21a21334 100644
--- a/lib/internal/streams/utils.js
+++ b/lib/internal/streams/utils.js
@@ -27,6 +27,15 @@ function isWritableNodeStream(obj) {
   );
 }
 
+function isDuplexNodeStream(obj) {
+  return !!(
+    obj &&
+    typeof obj.pipe === 'function' &&
+    typeof obj.on === 'function' &&
+    typeof obj.write === 'function'
+  );
+}
+
 function isNodeStream(obj) {
   return (
     obj &&
@@ -199,6 +208,7 @@ module.exports = {
   kDestroyed,
   isClosed,
   isDestroyed,
+  isDuplexNodeStream,
   isFinished,
   isIterable,
   isReadable,

From b633fd99a7081e4449369f7994f8782b3e391405 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 26 Jul 2021 15:26:14 +0200
Subject: [PATCH 02/18] Update lib/internal/streams/duplexify.js

Co-authored-by: Antoine du Hamel <duhamelantoine1995@gmail.com>
---
 lib/internal/streams/duplexify.js | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js
index 7c71824bf76844..8ec584163a9e29 100644
--- a/lib/internal/streams/duplexify.js
+++ b/lib/internal/streams/duplexify.js
@@ -88,7 +88,7 @@ module.exports = function duplexify(body, name) {
       });
     }
 
-    const { then } = value;
+    const then = value?.then;
     if (typeof then === 'function') {
       let d;
 

From 3ac44d0289f1d6dc7badae7ea4d2bf73749f3a7e Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 26 Jul 2021 15:39:17 +0200
Subject: [PATCH 03/18] fixup

---
 lib/internal/streams/destroy.js          |   2 +-
 lib/internal/streams/duplexify.js        |  88 +++++++++++------
 lib/internal/streams/end-of-stream.js    |  16 ++-
 lib/internal/streams/utils.js            |   6 +-
 test/parallel/test-stream-duplex-from.js | 120 +++++++++++++++++++++++
 5 files changed, 193 insertions(+), 39 deletions(-)
 create mode 100644 test/parallel/test-stream-duplex-from.js

diff --git a/lib/internal/streams/destroy.js b/lib/internal/streams/destroy.js
index 685ef90285048c..7d3657443e6ab5 100644
--- a/lib/internal/streams/destroy.js
+++ b/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;
   }
 
diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js
index 8ec584163a9e29..16e3bc21563495 100644
--- a/lib/internal/streams/duplexify.js
+++ b/lib/internal/streams/duplexify.js
@@ -66,17 +66,29 @@ class Duplexify extends Duplex {
 module.exports = function duplexify(body, name) {
   if (isDuplexNodeStream(body)) {
     return body;
-  } else if (isReadableNodeStream(body)) {
+  }
+
+  if (isReadableNodeStream(body)) {
     return _duplexify({ readable: body });
-  } else if (isWritableNodeStream(body)) {
+  }
+
+  if (isWritableNodeStream(body)) {
     return _duplexify({ writable: body });
-  } else if (isNodeStream(body)) {
+  }
+
+  if (isNodeStream(body)) {
     return _duplexify({ writable: false, readable: false });
-  } else if (isReadableStream(body)) {
+  }
+
+  if (isReadableStream(body)) {
     return _duplexify({ readable: Readable.fromWeb(body) });
-  } else if (isWritableStream(body)) {
+  }
+
+  if (isWritableStream(body)) {
     return _duplexify({ writable: Writable.fromWeb(body) });
-  } else if (typeof body === 'function') {
+  }
+
+  if (typeof body === 'function') {
     const { value, write, final } = fromAsyncGen(body);
 
     if (isIterable(value)) {
@@ -118,39 +130,50 @@ module.exports = function duplexify(body, name) {
 
     throw new ERR_INVALID_RETURN_VALUE(
       'Iterable, AsyncIterable or AsyncFunction', name, value);
-  } else if (isBlob(body)) {
+  }
+
+  if (isBlob(body)) {
     return duplexify(body.arrayBuffer());
-  } else if (isIterable(body)) {
+  }
+
+  if (isIterable(body)) {
     return from(Duplexify, body, {
       // TODO (ronag): highWaterMark?
       objectMode: true,
       writable: false
     });
-  } else if (
+  }
+
+  if (
     isReadableStream(body?.readable) &&
     isWritableStream(body?.writable)
   ) {
     return Duplexify.fromWeb(body);
-  } else if (
+  }
+
+  if (
     typeof body?.writable === 'object' ||
     typeof body?.readable === 'object'
   ) {
     const readable = body?.readable ?
       isReadableNodeStream(body?.readable) ? body?.readable :
-        duplexify(body.readable, body.options ?? body.readableOptions) :
+        duplexify(body.readable) :
       undefined;
 
     const writable = body?.writable ?
       isWritableNodeStream(body?.writable) ? body?.writable :
-        duplexify(body.writable, body.options ?? body.writableOptions) :
+        duplexify(body.writable) :
       undefined;
 
     return _duplexify({ readable, writable });
-  } else if (typeof body?.then === 'function') {
+  }
+
+  const then = body?.then;
+  if (typeof then === 'function') {
     let d;
 
     FunctionPrototypeCall(
-      body.then,
+      then,
       body,
       (val) => {
         if (val != null) {
@@ -165,7 +188,8 @@ module.exports = function duplexify(body, name) {
 
     return d = new Duplexify({
       objectMode: true,
-      writable: false
+      writable: false,
+      read() {}
     });
   }
 
@@ -200,7 +224,7 @@ function fromAsyncGen(fn) {
 }
 
 function _duplexify(pair) {
-  const r = typeof pair.readable.read !== 'function' ?
+  const r = pair.readable && typeof pair.readable.read !== 'function' ?
     Readable.wrap(pair.readable) : pair.readable;
   const w = pair.writable;
 
@@ -226,22 +250,6 @@ function _duplexify(pair) {
     }
   }
 
-  eos(r, (err) => {
-    readable = false;
-    if (err) {
-      destroyer(w, err);
-    }
-    onfinished(err);
-  });
-
-  eos(w, (err) => {
-    writable = false;
-    if (err) {
-      destroyer(r, err);
-    }
-    onfinished(err);
-  });
-
   // TODO(ronag): Avoid double buffering.
   // Implement Writable/Readable/Duplex traits.
   // See, https://github.com/nodejs/node/pull/33515.
@@ -254,6 +262,14 @@ function _duplexify(pair) {
   });
 
   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();
@@ -285,6 +301,14 @@ function _duplexify(pair) {
   }
 
   if (readable) {
+    eos(r, (err) => {
+      readable = false;
+      if (err) {
+        destroyer(r, err);
+      }
+      onfinished(err);
+    });
+
     r.on('readable', function() {
       if (onreadable) {
         const cb = onreadable;
diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index 3c1f0b27971927..7ca87990fff0ff 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -17,10 +17,17 @@ const {
   validateObject,
 } = require('internal/validators');
 
+const {
+  isBrandCheck,
+} = require('internal/webstreams/util');
+
+const isReadableStream =
+  isBrandCheck('ReadableStream');
+const isWritableStream =
+  isBrandCheck('WritableStream');
+
 const {
   isClosed,
-  isReadableStream,
-  isWritableStream,
   isReadable,
   isReadableNodeStream,
   isReadableFinished,
@@ -31,7 +38,7 @@ const {
   willEmitClose: _willEmitClose,
 } = require('internal/streams/utils');
 
-const Duplex = require('./duplex');
+let Duplex;
 
 function isRequest(stream) {
   return stream.setHeader && typeof stream.abort === 'function';
@@ -61,6 +68,9 @@ function eos(stream, options, callback) {
   if (isNodeStream(stream)) {
     // Do nothing...
   } if (isReadableStream(stream) || isWritableStream(stream)) {
+    if (!Duplex) {
+      Duplex = require('./duplex');
+    }
     stream = Duplex.from(stream);
   } else {
     // TODO: Throw INVALID_ARG_TYPE.
diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js
index e6543c21a21334..5d6ca481015a7a 100644
--- a/lib/internal/streams/utils.js
+++ b/lib/internal/streams/utils.js
@@ -30,7 +30,7 @@ function isWritableNodeStream(obj) {
 function isDuplexNodeStream(obj) {
   return !!(
     obj &&
-    typeof obj.pipe === 'function' &&
+    (typeof obj.pipe === 'function' && obj._readableState) &&
     typeof obj.on === 'function' &&
     typeof obj.write === 'function'
   );
@@ -111,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);
 }
diff --git a/test/parallel/test-stream-duplex-from.js b/test/parallel/test-stream-duplex-from.js
new file mode 100644
index 00000000000000..ae906177fcf317
--- /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, finished } = 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);
+  }));
+}

From cdaef9639e4dc2738d111587a70547f3675294b6 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Tue, 27 Jul 2021 12:48:32 +0200
Subject: [PATCH 04/18] fixup

---
 test/parallel/test-stream-duplex-from.js | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/test/parallel/test-stream-duplex-from.js b/test/parallel/test-stream-duplex-from.js
index ae906177fcf317..265b61dfd062f9 100644
--- a/test/parallel/test-stream-duplex-from.js
+++ b/test/parallel/test-stream-duplex-from.js
@@ -2,7 +2,7 @@
 
 const common = require('../common');
 const assert = require('assert');
-const { Duplex, Readable, Writable, finished } = require('stream');
+const { Duplex, Readable, Writable } = require('stream');
 
 {
   const d = Duplex.from({

From a6ad91e6504d5b348dc3fb55eb1b73d6527cd7ef Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Wed, 28 Jul 2021 08:25:26 +0200
Subject: [PATCH 05/18] fixup

---
 lib/internal/streams/compose.js | 24 +++++++++++++++++++++++-
 1 file changed, 23 insertions(+), 1 deletion(-)

diff --git a/lib/internal/streams/compose.js b/lib/internal/streams/compose.js
index 65a36e17710935..04b6c6dcb0a53e 100644
--- a/lib/internal/streams/compose.js
+++ b/lib/internal/streams/compose.js
@@ -16,6 +16,28 @@ const {
   },
 } = require('internal/errors');
 
+// 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');
@@ -85,7 +107,7 @@ 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 Duplex({
+  d = new ComposeDuplex({
     // TODO (ronag): highWaterMark?
     writableObjectMode: !!head?.writableObjectMode,
     readableObjectMode: !!tail?.writableObjectMode,

From 852d1648dec56c68b6799fb27be61dc03eda2345 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Fri, 30 Jul 2021 08:07:21 +0200
Subject: [PATCH 06/18] fixup: escape promise

---
 lib/internal/streams/duplexify.js | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js
index 16e3bc21563495..e0da22bb2bb6b4 100644
--- a/lib/internal/streams/duplexify.js
+++ b/lib/internal/streams/duplexify.js
@@ -123,7 +123,14 @@ module.exports = function duplexify(body, name) {
         readable: false,
         write,
         final(cb) {
-          final(() => promise.then(cb, cb));
+          final(() => promise.then(
+            function() {
+              process.nextTick(cb, null);
+            },
+            function(err) {
+              process.nextTick(cb, err);
+            })
+          );
         }
       });
     }

From ca32e5ab576e250d7e102574b798dd03bf421c0c Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Fri, 30 Jul 2021 08:21:53 +0200
Subject: [PATCH 07/18] fixup: remove eos webstream changes

---
 lib/internal/streams/end-of-stream.js | 23 -----------------------
 1 file changed, 23 deletions(-)

diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index 7ca87990fff0ff..274c2796edd443 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -17,15 +17,6 @@ const {
   validateObject,
 } = require('internal/validators');
 
-const {
-  isBrandCheck,
-} = require('internal/webstreams/util');
-
-const isReadableStream =
-  isBrandCheck('ReadableStream');
-const isWritableStream =
-  isBrandCheck('WritableStream');
-
 const {
   isClosed,
   isReadable,
@@ -34,12 +25,9 @@ const {
   isWritable,
   isWritableNodeStream,
   isWritableFinished,
-  isNodeStream,
   willEmitClose: _willEmitClose,
 } = require('internal/streams/utils');
 
-let Duplex;
-
 function isRequest(stream) {
   return stream.setHeader && typeof stream.abort === 'function';
 }
@@ -65,17 +53,6 @@ function eos(stream, options, callback) {
   const writable = options.writable ||
     (options.writable !== false && isWritableNodeStream(stream));
 
-  if (isNodeStream(stream)) {
-    // Do nothing...
-  } if (isReadableStream(stream) || isWritableStream(stream)) {
-    if (!Duplex) {
-      Duplex = require('./duplex');
-    }
-    stream = Duplex.from(stream);
-  } else {
-    // TODO: Throw INVALID_ARG_TYPE.
-  }
-
   const wState = stream._writableState;
   const rState = stream._readableState;
 

From 6d06b1c78465bde5b45915ddf7428df411c04fb2 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Fri, 30 Jul 2021 09:57:16 +0200
Subject: [PATCH 08/18] Revert "fixup: remove eos webstream changes"

This reverts commit ca32e5ab576e250d7e102574b798dd03bf421c0c.
---
 lib/internal/streams/end-of-stream.js | 23 +++++++++++++++++++++++
 1 file changed, 23 insertions(+)

diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index 274c2796edd443..7ca87990fff0ff 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -17,6 +17,15 @@ const {
   validateObject,
 } = require('internal/validators');
 
+const {
+  isBrandCheck,
+} = require('internal/webstreams/util');
+
+const isReadableStream =
+  isBrandCheck('ReadableStream');
+const isWritableStream =
+  isBrandCheck('WritableStream');
+
 const {
   isClosed,
   isReadable,
@@ -25,9 +34,12 @@ const {
   isWritable,
   isWritableNodeStream,
   isWritableFinished,
+  isNodeStream,
   willEmitClose: _willEmitClose,
 } = require('internal/streams/utils');
 
+let Duplex;
+
 function isRequest(stream) {
   return stream.setHeader && typeof stream.abort === 'function';
 }
@@ -53,6 +65,17 @@ function eos(stream, options, callback) {
   const writable = options.writable ||
     (options.writable !== false && isWritableNodeStream(stream));
 
+  if (isNodeStream(stream)) {
+    // Do nothing...
+  } if (isReadableStream(stream) || isWritableStream(stream)) {
+    if (!Duplex) {
+      Duplex = require('./duplex');
+    }
+    stream = Duplex.from(stream);
+  } else {
+    // TODO: Throw INVALID_ARG_TYPE.
+  }
+
   const wState = stream._writableState;
   const rState = stream._readableState;
 

From 95d043675f0079cef22281233aaaaadf17184489 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Fri, 30 Jul 2021 09:58:17 +0200
Subject: [PATCH 09/18] fixup: remove websreams + todo

---
 lib/internal/streams/duplexify.js     | 37 +++++++++++----------------
 lib/internal/streams/end-of-stream.js | 17 +-----------
 2 files changed, 16 insertions(+), 38 deletions(-)

diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js
index e0da22bb2bb6b4..731d37c20a2506 100644
--- a/lib/internal/streams/duplexify.js
+++ b/lib/internal/streams/duplexify.js
@@ -20,7 +20,6 @@ const {
 const { destroyer } = require('internal/streams/destroy');
 const Duplex = require('internal/streams/duplex');
 const Readable = require('internal/streams/readable');
-const Writable = require('internal/streams/writable');
 const { createDeferredPromise } = require('internal/util');
 const from = require('internal/streams/from');
 
@@ -28,15 +27,6 @@ const {
   isBlob,
 } = require('internal/blob');
 
-const {
-  isBrandCheck,
-} = require('internal/webstreams/util');
-
-const isReadableStream =
-  isBrandCheck('ReadableStream');
-const isWritableStream =
-  isBrandCheck('WritableStream');
-
 const {
   FunctionPrototypeCall
 } = primordials;
@@ -80,13 +70,15 @@ module.exports = function duplexify(body, name) {
     return _duplexify({ writable: false, readable: false });
   }
 
-  if (isReadableStream(body)) {
-    return _duplexify({ readable: Readable.fromWeb(body) });
-  }
+  // TODO: Webstreams
+  // if (isReadableStream(body)) {
+  //   return _duplexify({ readable: Readable.fromWeb(body) });
+  // }
 
-  if (isWritableStream(body)) {
-    return _duplexify({ writable: Writable.fromWeb(body) });
-  }
+  // TODO: Webstreams
+  // if (isWritableStream(body)) {
+  //   return _duplexify({ writable: Writable.fromWeb(body) });
+  // }
 
   if (typeof body === 'function') {
     const { value, write, final } = fromAsyncGen(body);
@@ -151,12 +143,13 @@ module.exports = function duplexify(body, name) {
     });
   }
 
-  if (
-    isReadableStream(body?.readable) &&
-    isWritableStream(body?.writable)
-  ) {
-    return Duplexify.fromWeb(body);
-  }
+  // TODO: Webstreams.
+  // if (
+  //   isReadableStream(body?.readable) &&
+  //   isWritableStream(body?.writable)
+  // ) {
+  //   return Duplexify.fromWeb(body);
+  // }
 
   if (
     typeof body?.writable === 'object' ||
diff --git a/lib/internal/streams/end-of-stream.js b/lib/internal/streams/end-of-stream.js
index 7ca87990fff0ff..c224b44eac6631 100644
--- a/lib/internal/streams/end-of-stream.js
+++ b/lib/internal/streams/end-of-stream.js
@@ -17,15 +17,6 @@ const {
   validateObject,
 } = require('internal/validators');
 
-const {
-  isBrandCheck,
-} = require('internal/webstreams/util');
-
-const isReadableStream =
-  isBrandCheck('ReadableStream');
-const isWritableStream =
-  isBrandCheck('WritableStream');
-
 const {
   isClosed,
   isReadable,
@@ -38,8 +29,6 @@ const {
   willEmitClose: _willEmitClose,
 } = require('internal/streams/utils');
 
-let Duplex;
-
 function isRequest(stream) {
   return stream.setHeader && typeof stream.abort === 'function';
 }
@@ -67,12 +56,8 @@ function eos(stream, options, callback) {
 
   if (isNodeStream(stream)) {
     // Do nothing...
-  } if (isReadableStream(stream) || isWritableStream(stream)) {
-    if (!Duplex) {
-      Duplex = require('./duplex');
-    }
-    stream = Duplex.from(stream);
   } else {
+    // TODO: Webstreams.
     // TODO: Throw INVALID_ARG_TYPE.
   }
 

From ad17b4155d03243460823dfe8cb437f8d2c4b992 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sun, 1 Aug 2021 12:34:15 +0200
Subject: [PATCH 10/18] Update lib/internal/streams/duplexify.js

Co-authored-by: Antoine du Hamel <duhamelantoine1995@gmail.com>
---
 lib/internal/streams/duplexify.js | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js
index 731d37c20a2506..0620d4840a7b34 100644
--- a/lib/internal/streams/duplexify.js
+++ b/lib/internal/streams/duplexify.js
@@ -115,14 +115,14 @@ module.exports = function duplexify(body, name) {
         readable: false,
         write,
         final(cb) {
-          final(() => promise.then(
-            function() {
+          final(aync () => {
+            try {
+              await promise;
               process.nextTick(cb, null);
-            },
-            function(err) {
+            } catch (err) {
               process.nextTick(cb, err);
-            })
-          );
+            }
+          });
         }
       });
     }

From b48f029f6ef8ecf7f1b106a859fcad73ef151161 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sun, 1 Aug 2021 14:10:09 +0200
Subject: [PATCH 11/18] ifuxp

---
 lib/internal/streams/duplexify.js | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/internal/streams/duplexify.js b/lib/internal/streams/duplexify.js
index 0620d4840a7b34..410a01df56126e 100644
--- a/lib/internal/streams/duplexify.js
+++ b/lib/internal/streams/duplexify.js
@@ -115,7 +115,7 @@ module.exports = function duplexify(body, name) {
         readable: false,
         write,
         final(cb) {
-          final(aync () => {
+          final(async () => {
             try {
               await promise;
               process.nextTick(cb, null);

From d214b6d9cb95c13347796c43c0a61ebdaffa552a Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 2 Aug 2021 12:46:56 +0200
Subject: [PATCH 12/18] fixup: docs

---
 doc/api/stream.md | 28 ++++++++++++++++++++++++++++
 1 file changed, 28 insertions(+)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index d616bb78bccfd0..a5903c153c8422 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -2031,6 +2031,34 @@ added: REPLACEME
 * `streamWritable` {stream.Writable}
 * Returns: {WritableStream}
 
+### `stream.Duplex.from(src)`
+<!-- YAML
+added:
+  - v12.3.0
+  - v10.17.0
+-->
+
+* `src` {{Stream|Blob|ArrayBuffer|string|Iterable|AsyncIterable|AsyncGeneratorFunction|AsyncFunction|Promise|WritableReadablePair}
+
+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`
+* `WritableReadablePair ({ 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.
+
 ### `stream.Duplex.fromWeb(pair[, options])`
 <!-- YAML
 added: REPLACEME

From c53b4517ceeb30ed442443b10847c6775af998f5 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 2 Aug 2021 12:52:13 +0200
Subject: [PATCH 13/18] Update doc/api/stream.md
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: Michaël Zasso <targos@protonmail.com>
---
 doc/api/stream.md | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index a5903c153c8422..c331c3a0cef685 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -2033,9 +2033,7 @@ added: REPLACEME
 
 ### `stream.Duplex.from(src)`
 <!-- YAML
-added:
-  - v12.3.0
-  - v10.17.0
+added: REPLACEME
 -->
 
 * `src` {{Stream|Blob|ArrayBuffer|string|Iterable|AsyncIterable|AsyncGeneratorFunction|AsyncFunction|Promise|WritableReadablePair}

From 0eb3a7282c6943a4336e91637d2bd0be01b6f408 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 2 Aug 2021 15:30:38 +0200
Subject: [PATCH 14/18] fixup

---
 doc/api/stream.md | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index c331c3a0cef685..f1c8b1eca6c658 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -2047,14 +2047,14 @@ A utility method for creating duplex streams.
 * `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
+* `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`
-* `WritableReadablePair ({ 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`.
+* `WritableReadablePair ({ 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.
 
 ### `stream.Duplex.fromWeb(pair[, options])`

From 680865a96743220fee6502f8dbfc2f95f2afa279 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 2 Aug 2021 16:28:47 +0200
Subject: [PATCH 15/18] fixup

---
 doc/api/stream.md | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index f1c8b1eca6c658..63ff58e0e09ad2 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -2036,7 +2036,8 @@ added: REPLACEME
 added: REPLACEME
 -->
 
-* `src` {{Stream|Blob|ArrayBuffer|string|Iterable|AsyncIterable|AsyncGeneratorFunction|AsyncFunction|Promise|WritableReadablePair}
+* `src` {{Stream|Blob|ArrayBuffer|string|Iterable|AsyncIterable|
+  AsyncGeneratorFunction|AsyncFunction|Promise|WritableReadablePair}
 
 A utility method for creating duplex streams.
 

From 8c760c6b2a1ea3aa8ef4fdaa315b548cb2ab970a Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 2 Aug 2021 16:29:40 +0200
Subject: [PATCH 16/18] fixup

---
 doc/api/stream.md | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 63ff58e0e09ad2..890c195494740d 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -2036,7 +2036,7 @@ added: REPLACEME
 added: REPLACEME
 -->
 
-* `src` {{Stream|Blob|ArrayBuffer|string|Iterable|AsyncIterable|
+* `src` {Stream|Blob|ArrayBuffer|string|Iterable|AsyncIterable|
   AsyncGeneratorFunction|AsyncFunction|Promise|WritableReadablePair}
 
 A utility method for creating duplex streams.
@@ -2057,6 +2057,7 @@ A utility method for creating duplex streams.
   `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

From 32b81cf097fce26ca59c78f75e45f9ac3659ad96 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Tue, 3 Aug 2021 08:53:47 +0200
Subject: [PATCH 17/18] fixup: type-parsers

---
 doc/api/stream.md         | 4 ++--
 tools/doc/type-parser.mjs | 2 ++
 2 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 890c195494740d..099d71badd8642 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -2037,7 +2037,7 @@ added: REPLACEME
 -->
 
 * `src` {Stream|Blob|ArrayBuffer|string|Iterable|AsyncIterable|
-  AsyncGeneratorFunction|AsyncFunction|Promise|WritableReadablePair}
+  AsyncGeneratorFunction|AsyncFunction|Promise|Object}
 
 A utility method for creating duplex streams.
 
@@ -2053,7 +2053,7 @@ A utility method for creating duplex streams.
   `null`.
 * `AsyncFunction` converts into a writable `Duplex`. Must return
   either `null` or `undefined`
-* `WritableReadablePair ({ writable, readable })` converts `readable` and
+* `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.
diff --git a/tools/doc/type-parser.mjs b/tools/doc/type-parser.mjs
index 481a35ca330c80..ed76c1afa0c200 100644
--- a/tools/doc/type-parser.mjs
+++ b/tools/doc/type-parser.mjs
@@ -34,6 +34,8 @@ const customTypesMap = {
 
   'AsyncIterable': 'https://tc39.github.io/ecma262/#sec-asynciterable-interface',
 
+  'AsyncGeneratorFunction':  'https://tc39.es/proposal-async-iteration/#sec-asyncgeneratorfunction-constructor',
+
   'bigint': `${jsDocPrefix}Reference/Global_Objects/BigInt`,
   'WebAssembly.Instance':
     `${jsDocPrefix}Reference/Global_Objects/WebAssembly/Instance`,

From 44d5004e2c136a5d27def9596ab0cbdccdd1be2c Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Tue, 3 Aug 2021 09:17:50 +0200
Subject: [PATCH 18/18] fixup: type-parser

---
 tools/doc/type-parser.mjs | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/tools/doc/type-parser.mjs b/tools/doc/type-parser.mjs
index ed76c1afa0c200..fb396070266831 100644
--- a/tools/doc/type-parser.mjs
+++ b/tools/doc/type-parser.mjs
@@ -34,7 +34,9 @@ const customTypesMap = {
 
   'AsyncIterable': 'https://tc39.github.io/ecma262/#sec-asynciterable-interface',
 
-  'AsyncGeneratorFunction':  'https://tc39.es/proposal-async-iteration/#sec-asyncgeneratorfunction-constructor',
+  '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':