From 4918a155dd2d2fa4f3bf87e5bf98ea7deabddaf6 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Fri, 18 Jun 2021 08:08:50 +0200
Subject: [PATCH 1/8] stream: add stream.compose

Refs: #32020
---
 doc/api/stream.md                       |  87 +++++
 lib/internal/streams/compose.js         | 288 ++++++++++++++++
 lib/internal/streams/pipeline.js        |   3 -
 lib/stream.js                           |   2 +
 test/parallel/test-bootstrap-modules.js |   1 +
 test/parallel/test-stream-compose.js    | 420 ++++++++++++++++++++++++
 6 files changed, 798 insertions(+), 3 deletions(-)
 create mode 100644 lib/internal/streams/compose.js
 create mode 100644 test/parallel/test-stream-compose.js

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 603e0407a8c60c..843f59c51d17fe 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -1859,6 +1859,93 @@ run().catch(console.error);
 after the `callback` has been invoked. In the case of reuse of streams after
 failure, this can cause event listener leaks and swallowed errors.
 
+### `stream.compose(...streams)`
+<!-- YAML
+added: REPLACEME
+-->
+
+* `streams` {Stream[]|Iterable[]|AsyncIterable[]|Function[]}
+* Returns: {stream.Duplex}
+
+Combines two or more streams into a `Duplex` stream that writes to the
+first stream and reads from the last. Each provided stream is piped into
+the next, using `stream.pipeline`. If any of the streams error then all
+are destroyed, including the outer `Duplex` stream.
+
+Because `stream.compose` returns a new stream that in turn can (and
+should) be piped into other streams, it enables composition. In contrast,
+when passing streams to `stream.pipeline`, typically the first stream is
+a readable stream and the last a writable stream, forming a closed
+circuit.
+
+If passed a `Function` is must be a factory method taking a `source`
+`Iterable`.
+
+```mjs
+import { compose, Transform } from 'stream';
+
+const removeSpaces = new Transform({
+  transform(chunk, encoding, callback) {
+    callback(null, String(chunk).replace(' ', ''));
+  }
+});
+
+async function* toUpper(source) {
+  for await (const chunk of source) {
+    yield String(chunk).toUpperCase();
+  }
+}
+
+let res = '';
+for await (const buf of compose(removeSpaces, toUpper).end('hello world')) {
+  res += buf;
+}
+
+console.log(res); // prints 'HELLOWORLD'
+```
+
+`stream.compose` can be used to convert async iterables, generators and
+functions into streams.
+
+* `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`.
+
+```mjs
+import { compose } from 'stream';
+import { finished } from 'stream/promises';
+
+// Convert AsyncIterable into readable Duplex.
+const s1 = compose(async function*() {
+  yield 'Hello';
+  yield 'World';
+}());
+
+// Convert AsyncGenerator into transform Duplex.
+const s2 = compose(async function*(source) {
+  for await (const chunk of source) {
+    yield String(chunk).toUpperCase();
+  }
+});
+
+let res = '';
+
+// Convert AsyncFunction into writable Duplex.
+const s3 = compose(async function(source) {
+  for await (const chunk of source) {
+    res += chunk;
+  }
+});
+
+await finished(compose(s1, s2, s3));
+
+console.log(res); // prints 'HELLOWORLD'
+```
+
 ### `stream.Readable.from(iterable, [options])`
 <!-- YAML
 added:
diff --git a/lib/internal/streams/compose.js b/lib/internal/streams/compose.js
new file mode 100644
index 00000000000000..8e1e16a3ce8574
--- /dev/null
+++ b/lib/internal/streams/compose.js
@@ -0,0 +1,288 @@
+'use strict';
+
+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 {
+  isStream,
+  isIterable,
+  isReadable,
+  isWritable,
+} = require('internal/streams/utils');
+const {
+  PromiseResolve,
+} = primordials;
+const {
+  AbortError,
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+    ERR_INVALID_ARG_VALUE,
+    ERR_INVALID_RETURN_VALUE,
+    ERR_MISSING_ARGS,
+  },
+} = require('internal/errors');
+const assert = require('internal/assert');
+
+// This is needed for pre node 17.
+class ComposeDuplex extends Duplex {
+  constructor(options) {
+    super(options);
+
+    // https://github.com/nodejs/node/pull/34385
+
+    if (options?.readable === false) {
+      this._readableState.readable = false;
+      this._readableState.ended = true;
+      this._readableState.endEmitted = true;
+    }
+
+    if (options?.writable === false) {
+      this._writableState.writable = false;
+      this._writableState.ending = true;
+      this._writableState.ended = true;
+      this._writableState.finished = true;
+    }
+  }
+}
+
+module.exports = function compose(...streams) {
+  if (streams.length === 0) {
+    throw new ERR_MISSING_ARGS('streams');
+  }
+
+  if (streams.length === 1) {
+    return makeDuplex(streams[0], 'streams[0]');
+  }
+
+  const orgStreams = [...streams];
+
+  if (typeof streams[0] === 'function') {
+    streams[0] = makeDuplex(streams[0], 'streams[0]');
+  }
+
+  if (typeof streams[streams.length - 1] === 'function') {
+    const idx = streams.length - 1;
+    streams[idx] = makeDuplex(streams[idx], `streams[${idx}]`);
+  }
+
+  for (let n = 0; n < streams.length; ++n) {
+    if (!isStream(streams[n])) {
+      // TODO (ronag): Add checks for non streams.
+      continue;
+    }
+    if (n < streams.length - 1 && !isReadable(streams[n])) {
+      throw new ERR_INVALID_ARG_VALUE(
+        `streams[${n}]`,
+        orgStreams[n],
+        'must be readable'
+      );
+    }
+    if (n > 0 && !isWritable(streams[n])) {
+      throw new ERR_INVALID_ARG_VALUE(
+        `streams[${n}]`,
+        orgStreams[n],
+        'must be writable'
+      );
+    }
+  }
+
+  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);
+    }
+  }
+
+  const head = streams[0];
+  const tail = pipeline(streams, onfinished);
+
+  const writable = isWritable(head);
+  const readable = isReadable(tail);
+
+  // TODO (ronag): Avoid double buffering.
+  // Implement Writable/Readable/Duplex traits.
+  // See, https://github.com/nodejs/node/pull/33515.
+  d = new ComposeDuplex({
+    highWaterMark: 1,
+    writableObjectMode: !!head?.writableObjectMode,
+    readableObjectMode: !!tail?.writableObjectMode,
+    writable,
+    readable,
+  });
+
+  if (writable) {
+    d._write = function(chunk, encoding, callback) {
+      if (head.write(chunk, encoding)) {
+        callback();
+      } else {
+        ondrain = callback;
+      }
+    };
+
+    d._final = function(callback) {
+      head.end();
+      onfinish = callback;
+    };
+
+    head.on('drain', function() {
+      if (ondrain) {
+        const cb = ondrain;
+        ondrain = null;
+        cb();
+      }
+    });
+
+    tail.on('finish', function() {
+      if (onfinish) {
+        const cb = onfinish;
+        onfinish = null;
+        cb();
+      }
+    });
+  }
+
+  if (readable) {
+    tail.on('readable', function() {
+      if (onreadable) {
+        const cb = onreadable;
+        onreadable = null;
+        cb();
+      }
+    });
+
+    tail.on('end', function() {
+      d.push(null);
+    });
+
+    d._read = function() {
+      while (true) {
+        const buf = tail.read();
+
+        if (buf === null) {
+          onreadable = d._read;
+          return;
+        }
+
+        if (!d.push(buf)) {
+          return;
+        }
+      }
+    };
+  }
+
+  if (!readable && !writable) {
+    tail.on('finish', function() {
+      d.destroy();
+    });
+  }
+
+  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(tail, err);
+    }
+  };
+
+  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 (isStream(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/pipeline.js b/lib/internal/streams/pipeline.js
index c98b3b3d21b633..1b56c08b9e6958 100644
--- a/lib/internal/streams/pipeline.js
+++ b/lib/internal/streams/pipeline.js
@@ -295,9 +295,6 @@ function pipeline(...streams) {
     }
   }
 
-  // TODO(ronag): Consider returning a Duplex proxy if the first argument
-  // is a writable. Would improve composability.
-  // See, https://github.com/nodejs/node/issues/32020
   return ret;
 }
 
diff --git a/lib/stream.js b/lib/stream.js
index 16a2370232e360..b84efb0fd8862d 100644
--- a/lib/stream.js
+++ b/lib/stream.js
@@ -30,6 +30,7 @@ const {
 } = require('internal/util');
 
 const pipeline = require('internal/streams/pipeline');
+const compose = require('internal/streams/compose');
 const { destroyer } = require('internal/streams/destroy');
 const eos = require('internal/streams/end-of-stream');
 const internalBuffer = require('internal/buffer');
@@ -47,6 +48,7 @@ const { addAbortSignal } = require('internal/streams/add-abort-signal');
 Stream.addAbortSignal = addAbortSignal;
 Stream.finished = eos;
 Stream.destroy = destroyer;
+Stream.compose = compose;
 
 ObjectDefineProperty(Stream, 'promises', {
   configurable: true,
diff --git a/test/parallel/test-bootstrap-modules.js b/test/parallel/test-bootstrap-modules.js
index 0ca00f31adce8c..3f90571ea78078 100644
--- a/test/parallel/test-bootstrap-modules.js
+++ b/test/parallel/test-bootstrap-modules.js
@@ -108,6 +108,7 @@ const expectedModules = new Set([
   'NativeModule internal/streams/legacy',
   'NativeModule internal/streams/passthrough',
   'NativeModule internal/streams/pipeline',
+  'NativeModule internal/streams/compose',
   'NativeModule internal/streams/readable',
   'NativeModule internal/streams/state',
   'NativeModule internal/streams/transform',
diff --git a/test/parallel/test-stream-compose.js b/test/parallel/test-stream-compose.js
new file mode 100644
index 00000000000000..5d64f9869f8836
--- /dev/null
+++ b/test/parallel/test-stream-compose.js
@@ -0,0 +1,420 @@
+'use strict';
+
+const common = require('../common');
+const {
+  Readable,
+  Transform,
+  Writable,
+  finished,
+  compose,
+  PassThrough
+} = require('stream');
+const assert = require('assert');
+
+{
+  let res = '';
+  compose(
+    new Transform({
+      transform: common.mustCall((chunk, encoding, callback) => {
+        callback(null, chunk + chunk);
+      })
+    }),
+    new Transform({
+      transform: common.mustCall((chunk, encoding, callback) => {
+        callback(null, chunk.toString().toUpperCase());
+      })
+    })
+  )
+  .end('asd')
+  .on('data', common.mustCall((buf) => {
+    res += buf;
+  }))
+  .on('end', common.mustCall(() => {
+    assert.strictEqual(res, 'ASDASD');
+  }));
+}
+
+{
+  let res = '';
+  compose(
+    async function*(source) {
+      for await (const chunk of source) {
+        yield chunk + chunk;
+      }
+    },
+    async function*(source) {
+      for await (const chunk of source) {
+        yield chunk.toString().toUpperCase();
+      }
+    }
+  )
+  .end('asd')
+  .on('data', common.mustCall((buf) => {
+    res += buf;
+  }))
+  .on('end', common.mustCall(() => {
+    assert.strictEqual(res, 'ASDASD');
+  }));
+}
+
+{
+  let res = '';
+  compose(
+    async function*(source) {
+      for await (const chunk of source) {
+        yield chunk + chunk;
+      }
+    }
+  )
+  .end('asd')
+  .on('data', common.mustCall((buf) => {
+    res += buf;
+  }))
+  .on('end', common.mustCall(() => {
+    assert.strictEqual(res, 'asdasd');
+  }));
+}
+
+{
+  let res = '';
+  compose(
+    Readable.from(['asd']),
+    new Transform({
+      transform: common.mustCall((chunk, encoding, callback) => {
+        callback(null, chunk.toString().toUpperCase());
+      })
+    })
+  )
+  .on('data', common.mustCall((buf) => {
+    res += buf;
+  }))
+  .on('end', common.mustCall(() => {
+    assert.strictEqual(res, 'ASD');
+  }));
+}
+
+{
+  let res = '';
+  compose(
+    async function* () {
+      yield 'asd';
+    }(),
+    new Transform({
+      transform: common.mustCall((chunk, encoding, callback) => {
+        callback(null, chunk.toString().toUpperCase());
+      })
+    })
+  )
+  .on('data', common.mustCall((buf) => {
+    res += buf;
+  }))
+  .on('end', common.mustCall(() => {
+    assert.strictEqual(res, 'ASD');
+  }));
+}
+
+{
+  let res = '';
+  compose(
+    new Transform({
+      transform: common.mustCall((chunk, encoding, callback) => {
+        callback(null, chunk.toString().toUpperCase());
+      })
+    }),
+    async function*(source) {
+      for await (const chunk of source) {
+        yield chunk;
+      }
+    },
+    new Writable({
+      write: common.mustCall((chunk, encoding, callback) => {
+        res += chunk;
+        callback(null);
+      })
+    })
+  )
+  .end('asd')
+  .on('finish', common.mustCall(() => {
+    assert.strictEqual(res, 'ASD');
+  }));
+}
+
+{
+  let res = '';
+  compose(
+    new Transform({
+      transform: common.mustCall((chunk, encoding, callback) => {
+        callback(null, chunk.toString().toUpperCase());
+      })
+    }),
+    async function*(source) {
+      for await (const chunk of source) {
+        yield chunk;
+      }
+    },
+    async function(source) {
+      for await (const chunk of source) {
+        res += chunk;
+      }
+    }
+  )
+  .end('asd')
+  .on('finish', common.mustCall(() => {
+    assert.strictEqual(res, 'ASD');
+  }));
+}
+
+{
+  let res;
+  compose(
+    new Transform({
+      objectMode: true,
+      transform: common.mustCall((chunk, encoding, callback) => {
+        callback(null, { chunk });
+      })
+    }),
+    async function*(source) {
+      for await (const chunk of source) {
+        yield chunk;
+      }
+    },
+    new Transform({
+      objectMode: true,
+      transform: common.mustCall((chunk, encoding, callback) => {
+        callback(null, { chunk });
+      })
+    })
+  )
+  .end(true)
+  .on('data', common.mustCall((buf) => {
+    res = buf;
+  }))
+  .on('end', common.mustCall(() => {
+    assert.strictEqual(res.chunk.chunk, true);
+  }));
+}
+
+{
+  const _err = new Error('asd');
+  compose(
+    new Transform({
+      objectMode: true,
+      transform: common.mustCall((chunk, encoding, callback) => {
+        callback(_err);
+      })
+    }),
+    async function*(source) {
+      for await (const chunk of source) {
+        yield chunk;
+      }
+    },
+    new Transform({
+      objectMode: true,
+      transform: common.mustNotCall((chunk, encoding, callback) => {
+        callback(null, { chunk });
+      })
+    })
+  )
+  .end(true)
+  .on('data', common.mustNotCall())
+  .on('end', common.mustNotCall())
+  .on('error', (err) => {
+    assert.strictEqual(err, _err);
+  });
+}
+
+{
+  const _err = new Error('asd');
+  compose(
+    new Transform({
+      objectMode: true,
+      transform: common.mustCall((chunk, encoding, callback) => {
+        callback(null, chunk);
+      })
+    }),
+    async function*(source) {
+      let tmp = '';
+      for await (const chunk of source) {
+        tmp += chunk;
+        throw _err;
+      }
+      return tmp;
+    },
+    new Transform({
+      objectMode: true,
+      transform: common.mustNotCall((chunk, encoding, callback) => {
+        callback(null, { chunk });
+      })
+    })
+  )
+  .end(true)
+  .on('data', common.mustNotCall())
+  .on('end', common.mustNotCall())
+  .on('error', (err) => {
+    assert.strictEqual(err, _err);
+  });
+}
+
+{
+  let buf = '';
+
+  // Convert into readable Duplex.
+  const s1 = compose(async function* () {
+    yield 'Hello';
+    yield 'World';
+  }(), async function* (source) {
+    for await (const chunk of source) {
+      yield String(chunk).toUpperCase();
+    }
+  }, async function(source) {
+    for await (const chunk of source) {
+      buf += chunk;
+    }
+  });
+
+  finished(s1.resume(), common.mustCall((err) => {
+    assert(!err);
+    assert.strictEqual(buf, 'HELLOWORLD');
+  }));
+}
+
+{
+  let buf = '';
+  // Convert into transform duplex.
+  const s2 = compose(async function* (source) {
+    for await (const chunk of source) {
+      yield String(chunk).toUpperCase();
+    }
+  });
+  s2.end('helloworld');
+  s2.resume();
+  s2.on('data', (chunk) => {
+    buf += chunk;
+  });
+
+  finished(s2.resume(), common.mustCall((err) => {
+    assert(!err);
+    assert.strictEqual(buf, 'HELLOWORLD');
+  }));
+}
+
+{
+  let buf = '';
+
+  // Convert into readable Duplex.
+  const s1 = compose(async function* () {
+    yield 'Hello';
+    yield 'World';
+  }());
+
+  // Convert into transform duplex.
+  const s2 = compose(async function* (source) {
+    for await (const chunk of source) {
+      yield String(chunk).toUpperCase();
+    }
+  });
+
+  // Convert into writable duplex.
+  const s3 = compose(async function(source) {
+    for await (const chunk of source) {
+      buf += chunk;
+    }
+  });
+
+  const s4 = compose(s1, s2, s3);
+
+  finished(s4, common.mustCall((err) => {
+    assert(!err);
+    assert.strictEqual(buf, 'HELLOWORLD');
+  }));
+}
+
+{
+  let buf = '';
+
+  // Convert into readable Duplex.
+  const s1 = compose(async function* () {
+    yield 'Hello';
+    yield 'World';
+  }(), async function* (source) {
+    for await (const chunk of source) {
+      yield String(chunk).toUpperCase();
+    }
+  }, async function(source) {
+    for await (const chunk of source) {
+      buf += chunk;
+    }
+  });
+
+  finished(s1, common.mustCall((err) => {
+    assert(!err);
+    assert.strictEqual(buf, 'HELLOWORLD');
+  }));
+}
+
+{
+  try {
+    compose();
+  } catch (err) {
+    assert.strictEqual(err.code, 'ERR_MISSING_ARGS');
+  }
+}
+
+{
+  try {
+    compose(new Writable(), new PassThrough());
+  } catch (err) {
+    assert.strictEqual(err.code, 'ERR_INVALID_ARG_VALUE');
+  }
+}
+
+{
+  try {
+    compose(new PassThrough(), new Readable({ read() {} }), new PassThrough());
+  } catch (err) {
+    assert.strictEqual(err.code, 'ERR_INVALID_ARG_VALUE');
+  }
+}
+
+{
+  let buf = '';
+
+  // Convert into readable Duplex.
+  const s1 = compose(async function* () {
+    yield 'Hello';
+    yield 'World';
+  }(), async function* (source) {
+    for await (const chunk of source) {
+      yield String(chunk).toUpperCase();
+    }
+  }, async function(source) {
+    for await (const chunk of source) {
+      buf += chunk;
+    }
+    return buf;
+  });
+
+  finished(s1, common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_INVALID_RETURN_VALUE');
+  }));
+}
+
+{
+  let buf = '';
+
+  // Convert into readable Duplex.
+  const s1 = compose('HelloWorld', async function* (source) {
+    for await (const chunk of source) {
+      yield String(chunk).toUpperCase();
+    }
+  }, async function(source) {
+    for await (const chunk of source) {
+      buf += chunk;
+    }
+  });
+
+  finished(s1, common.mustCall((err) => {
+    assert(!err);
+    assert.strictEqual(buf, 'HELLOWORLD');
+  }));
+}

From d065a2bf596f3fe48b46c5ce12651561edfa1034 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 12 Jul 2021 12:31:01 +0200
Subject: [PATCH 2/8] fixup: remove async fn support

---
 doc/api/stream.md | 55 +++++------------------------------------------
 lib/stream.js     | 22 ++++++++++++++++++-
 2 files changed, 26 insertions(+), 51 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 843f59c51d17fe..5cb6f009dd3d59 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -1864,7 +1864,7 @@ failure, this can cause event listener leaks and swallowed errors.
 added: REPLACEME
 -->
 
-* `streams` {Stream[]|Iterable[]|AsyncIterable[]|Function[]}
+* `streams` {Stream[]}
 * Returns: {stream.Duplex}
 
 Combines two or more streams into a `Duplex` stream that writes to the
@@ -1878,9 +1878,6 @@ when passing streams to `stream.pipeline`, typically the first stream is
 a readable stream and the last a writable stream, forming a closed
 circuit.
 
-If passed a `Function` is must be a factory method taking a `source`
-`Iterable`.
-
 ```mjs
 import { compose, Transform } from 'stream';
 
@@ -1890,11 +1887,11 @@ const removeSpaces = new Transform({
   }
 });
 
-async function* toUpper(source) {
-  for await (const chunk of source) {
-    yield String(chunk).toUpperCase();
+const toUpper = new Transform({
+  transform(chunk, encoding, callback) {
+    callback(null, String(chunk).toUpperCase());
   }
-}
+});
 
 let res = '';
 for await (const buf of compose(removeSpaces, toUpper).end('hello world')) {
@@ -1904,48 +1901,6 @@ for await (const buf of compose(removeSpaces, toUpper).end('hello world')) {
 console.log(res); // prints 'HELLOWORLD'
 ```
 
-`stream.compose` can be used to convert async iterables, generators and
-functions into streams.
-
-* `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`.
-
-```mjs
-import { compose } from 'stream';
-import { finished } from 'stream/promises';
-
-// Convert AsyncIterable into readable Duplex.
-const s1 = compose(async function*() {
-  yield 'Hello';
-  yield 'World';
-}());
-
-// Convert AsyncGenerator into transform Duplex.
-const s2 = compose(async function*(source) {
-  for await (const chunk of source) {
-    yield String(chunk).toUpperCase();
-  }
-});
-
-let res = '';
-
-// Convert AsyncFunction into writable Duplex.
-const s3 = compose(async function(source) {
-  for await (const chunk of source) {
-    res += chunk;
-  }
-});
-
-await finished(compose(s1, s2, s3));
-
-console.log(res); // prints 'HELLOWORLD'
-```
-
 ### `stream.Readable.from(iterable, [options])`
 <!-- YAML
 added:
diff --git a/lib/stream.js b/lib/stream.js
index b84efb0fd8862d..b3a723781354cc 100644
--- a/lib/stream.js
+++ b/lib/stream.js
@@ -34,6 +34,12 @@ const compose = require('internal/streams/compose');
 const { destroyer } = require('internal/streams/destroy');
 const eos = require('internal/streams/end-of-stream');
 const internalBuffer = require('internal/buffer');
+const { isStream } = require('internal/streams/utils');
+const {
+  codes: {
+    ERR_INVALID_ARG_VALUE,
+  },
+} = require('internal/errors');
 
 const promises = require('stream/promises');
 
@@ -48,7 +54,21 @@ const { addAbortSignal } = require('internal/streams/add-abort-signal');
 Stream.addAbortSignal = addAbortSignal;
 Stream.finished = eos;
 Stream.destroy = destroyer;
-Stream.compose = compose;
+
+Stream.compose = function(...streams) {
+  // TODO (ronag): Remove this once async function API
+  // has been discussed.
+  for (let n = 0; n < streams.length; ++n) {
+    if (!isStream(streams[n])) {
+      throw new ERR_INVALID_ARG_VALUE(
+        `streams[${n}]`,
+        streams[n],
+        'must be stream'
+      );
+    }
+  }
+  return compose(...streams);
+};
 
 ObjectDefineProperty(Stream, 'promises', {
   configurable: true,

From 882b15bb79af15ec8055dfcdc3765af2abf3156b Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 12 Jul 2021 13:29:38 +0200
Subject: [PATCH 3/8] fixup

---
 lib/internal/streams/compose.js | 6 +++---
 lib/stream.js                   | 4 ++--
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/lib/internal/streams/compose.js b/lib/internal/streams/compose.js
index 8e1e16a3ce8574..82cd249d4e856c 100644
--- a/lib/internal/streams/compose.js
+++ b/lib/internal/streams/compose.js
@@ -6,7 +6,7 @@ const { createDeferredPromise } = require('internal/util');
 const { destroyer } = require('internal/streams/destroy');
 const from = require('internal/streams/from');
 const {
-  isStream,
+  isNodeStream,
   isIterable,
   isReadable,
   isWritable,
@@ -68,7 +68,7 @@ module.exports = function compose(...streams) {
   }
 
   for (let n = 0; n < streams.length; ++n) {
-    if (!isStream(streams[n])) {
+    if (!isNodeStream(streams[n])) {
       // TODO (ronag): Add checks for non streams.
       continue;
     }
@@ -246,7 +246,7 @@ function makeDuplex(stream, name) {
       throw new ERR_INVALID_RETURN_VALUE(
         'Iterable, AsyncIterable or AsyncFunction', name, value);
     }
-  } else if (isStream(stream)) {
+  } else if (isNodeStream(stream)) {
     ret = stream;
   } else if (isIterable(stream)) {
     ret = from(ComposeDuplex, stream, {
diff --git a/lib/stream.js b/lib/stream.js
index b3a723781354cc..332720e7bfb09f 100644
--- a/lib/stream.js
+++ b/lib/stream.js
@@ -34,7 +34,7 @@ const compose = require('internal/streams/compose');
 const { destroyer } = require('internal/streams/destroy');
 const eos = require('internal/streams/end-of-stream');
 const internalBuffer = require('internal/buffer');
-const { isStream } = require('internal/streams/utils');
+const { isNodeStream } = require('internal/streams/utils');
 const {
   codes: {
     ERR_INVALID_ARG_VALUE,
@@ -59,7 +59,7 @@ Stream.compose = function(...streams) {
   // TODO (ronag): Remove this once async function API
   // has been discussed.
   for (let n = 0; n < streams.length; ++n) {
-    if (!isStream(streams[n])) {
+    if (!isNodeStream(streams[n])) {
       throw new ERR_INVALID_ARG_VALUE(
         `streams[${n}]`,
         streams[n],

From 0cf9ee54c2e5d4e3257b6f17ed8a835bf5eceb5a Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 12 Jul 2021 14:13:39 +0200
Subject: [PATCH 4/8] fixup

---
 lib/internal/streams/compose.js      |  8 ++------
 lib/internal/streams/utils.js        | 10 +++++++++-
 test/parallel/test-stream-compose.js |  4 +++-
 3 files changed, 14 insertions(+), 8 deletions(-)

diff --git a/lib/internal/streams/compose.js b/lib/internal/streams/compose.js
index 82cd249d4e856c..e88603703b6395 100644
--- a/lib/internal/streams/compose.js
+++ b/lib/internal/streams/compose.js
@@ -102,6 +102,8 @@ module.exports = function compose(...streams) {
       cb(err);
     } else if (err) {
       d.destroy(err);
+    } else if (!readable && !writable) {
+      d.destroy();
     }
   }
 
@@ -182,12 +184,6 @@ module.exports = function compose(...streams) {
     };
   }
 
-  if (!readable && !writable) {
-    tail.on('finish', function() {
-      d.destroy();
-    });
-  }
-
   d._destroy = function(err, callback) {
     if (!err && onclose !== null) {
       err = new AbortError();
diff --git a/lib/internal/streams/utils.js b/lib/internal/streams/utils.js
index 0cd9c8eb36062f..01396b5113340f 100644
--- a/lib/internal/streams/utils.js
+++ b/lib/internal/streams/utils.js
@@ -28,7 +28,15 @@ function isWritableNodeStream(obj) {
 }
 
 function isNodeStream(obj) {
-  return isReadableNodeStream(obj) || isWritableNodeStream(obj);
+  return (
+    obj &&
+    (
+      obj._readableState ||
+      obj._writableState ||
+      (typeof obj.write === 'function' && typeof obj.on === 'function') ||
+      (typeof obj.pipe === 'function' && typeof obj.on === 'function')
+    )
+  );
 }
 
 function isIterable(obj, isAsync) {
diff --git a/test/parallel/test-stream-compose.js b/test/parallel/test-stream-compose.js
index 5d64f9869f8836..9b7a3e5112d35e 100644
--- a/test/parallel/test-stream-compose.js
+++ b/test/parallel/test-stream-compose.js
@@ -1,3 +1,5 @@
+// Flags: --expose-internals
+
 'use strict';
 
 const common = require('../common');
@@ -6,9 +8,9 @@ const {
   Transform,
   Writable,
   finished,
-  compose,
   PassThrough
 } = require('stream');
+const compose = require('internal/streams/compose');
 const assert = require('assert');
 
 {

From 9fa8a9f93314abf1707fecbefa67595f4d75acad Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 12 Jul 2021 14:18:20 +0200
Subject: [PATCH 5/8] fixup

---
 lib/internal/streams/compose.js      | 4 ++--
 test/parallel/test-stream-compose.js | 3 +++
 2 files changed, 5 insertions(+), 2 deletions(-)

diff --git a/lib/internal/streams/compose.js b/lib/internal/streams/compose.js
index e88603703b6395..c3210f6e3b14ca 100644
--- a/lib/internal/streams/compose.js
+++ b/lib/internal/streams/compose.js
@@ -110,8 +110,8 @@ module.exports = function compose(...streams) {
   const head = streams[0];
   const tail = pipeline(streams, onfinished);
 
-  const writable = isWritable(head);
-  const readable = isReadable(tail);
+  const writable = !!isWritable(head);
+  const readable = !!isReadable(tail);
 
   // TODO (ronag): Avoid double buffering.
   // Implement Writable/Readable/Duplex traits.
diff --git a/test/parallel/test-stream-compose.js b/test/parallel/test-stream-compose.js
index 9b7a3e5112d35e..c3d52e08e0048e 100644
--- a/test/parallel/test-stream-compose.js
+++ b/test/parallel/test-stream-compose.js
@@ -274,6 +274,9 @@ const assert = require('assert');
     }
   });
 
+  assert.strictEqual(s1.writable, false);
+  assert.strictEqual(s1.readable, false);
+
   finished(s1.resume(), common.mustCall((err) => {
     assert(!err);
     assert.strictEqual(buf, 'HELLOWORLD');

From 26bdf985d12f6ea06fdc4ec0568cd2d6a6185c09 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Mon, 12 Jul 2021 18:19:54 +0200
Subject: [PATCH 6/8] fixup: alpha order

---
 test/parallel/test-bootstrap-modules.js | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/test/parallel/test-bootstrap-modules.js b/test/parallel/test-bootstrap-modules.js
index 3f90571ea78078..6322df07dbe5c8 100644
--- a/test/parallel/test-bootstrap-modules.js
+++ b/test/parallel/test-bootstrap-modules.js
@@ -101,6 +101,7 @@ const expectedModules = new Set([
   'NativeModule internal/stream_base_commons',
   'NativeModule internal/streams/add-abort-signal',
   'NativeModule internal/streams/buffer_list',
+  'NativeModule internal/streams/compose',
   'NativeModule internal/streams/destroy',
   'NativeModule internal/streams/duplex',
   'NativeModule internal/streams/end-of-stream',
@@ -108,7 +109,6 @@ const expectedModules = new Set([
   'NativeModule internal/streams/legacy',
   'NativeModule internal/streams/passthrough',
   'NativeModule internal/streams/pipeline',
-  'NativeModule internal/streams/compose',
   'NativeModule internal/streams/readable',
   'NativeModule internal/streams/state',
   'NativeModule internal/streams/transform',

From 87302f01b8d372ddf5cc1588a09353f56d2cbe4f Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Tue, 13 Jul 2021 10:06:51 +0200
Subject: [PATCH 7/8] Apply suggestions from code review
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: Michaël Zasso <targos@protonmail.com>
---
 lib/internal/streams/compose.js | 4 ++--
 lib/stream.js                   | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/lib/internal/streams/compose.js b/lib/internal/streams/compose.js
index c3210f6e3b14ca..bdfdc4cebe0c47 100644
--- a/lib/internal/streams/compose.js
+++ b/lib/internal/streams/compose.js
@@ -69,7 +69,7 @@ module.exports = function compose(...streams) {
 
   for (let n = 0; n < streams.length; ++n) {
     if (!isNodeStream(streams[n])) {
-      // TODO (ronag): Add checks for non streams.
+      // TODO(ronag): Add checks for non streams.
       continue;
     }
     if (n < streams.length - 1 && !isReadable(streams[n])) {
@@ -113,7 +113,7 @@ module.exports = function compose(...streams) {
   const writable = !!isWritable(head);
   const readable = !!isReadable(tail);
 
-  // TODO (ronag): Avoid double buffering.
+  // TODO(ronag): Avoid double buffering.
   // Implement Writable/Readable/Duplex traits.
   // See, https://github.com/nodejs/node/pull/33515.
   d = new ComposeDuplex({
diff --git a/lib/stream.js b/lib/stream.js
index 332720e7bfb09f..342ea2b898906f 100644
--- a/lib/stream.js
+++ b/lib/stream.js
@@ -56,7 +56,7 @@ Stream.finished = eos;
 Stream.destroy = destroyer;
 
 Stream.compose = function(...streams) {
-  // TODO (ronag): Remove this once async function API
+  // TODO(ronag): Remove this once async function API
   // has been discussed.
   for (let n = 0; n < streams.length; ++n) {
     if (!isNodeStream(streams[n])) {

From dc80a2625df95d739b262b48b736715d164c320c Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Tue, 13 Jul 2021 10:07:01 +0200
Subject: [PATCH 8/8] fixup

---
 lib/stream.js | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/lib/stream.js b/lib/stream.js
index 342ea2b898906f..5201991bf1194a 100644
--- a/lib/stream.js
+++ b/lib/stream.js
@@ -30,7 +30,7 @@ const {
 } = require('internal/util');
 
 const pipeline = require('internal/streams/pipeline');
-const compose = require('internal/streams/compose');
+const _compose = require('internal/streams/compose');
 const { destroyer } = require('internal/streams/destroy');
 const eos = require('internal/streams/end-of-stream');
 const internalBuffer = require('internal/buffer');
@@ -55,8 +55,8 @@ Stream.addAbortSignal = addAbortSignal;
 Stream.finished = eos;
 Stream.destroy = destroyer;
 
-Stream.compose = function(...streams) {
-  // TODO(ronag): Remove this once async function API
+Stream.compose = function compose(...streams) {
+  // TODO (ronag): Remove this once async function API
   // has been discussed.
   for (let n = 0; n < streams.length; ++n) {
     if (!isNodeStream(streams[n])) {
@@ -67,7 +67,7 @@ Stream.compose = function(...streams) {
       );
     }
   }
-  return compose(...streams);
+  return _compose(...streams);
 };
 
 ObjectDefineProperty(Stream, 'promises', {