Upload Kmake

lib/internal/streams/add-abort-signal.js

@@ -0,0 +1,60 @@
+'use strict';
+
+const {
+  SymbolDispose,
+} = primordials;
+
+const {
+  AbortError,
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+  },
+} = require('internal/errors');
+
+const {
+  isNodeStream,
+  isWebStream,
+  kControllerErrorFunction,
+} = require('internal/streams/utils');
+
+const eos = require('internal/streams/end-of-stream');
+let addAbortListener;
+
+// This method is inlined here for readable-stream
+// It also does not allow for signal to not exist on the stream
+// https://github.com/nodejs/node/pull/36061#discussion_r533718029
+const validateAbortSignal = (signal, name) => {
+  if (typeof signal !== 'object' ||
+       !('aborted' in signal)) {
+    throw new ERR_INVALID_ARG_TYPE(name, 'AbortSignal', signal);
+  }
+};
+
+module.exports.addAbortSignal = function addAbortSignal(signal, stream) {
+  validateAbortSignal(signal, 'signal');
+  if (!isNodeStream(stream) && !isWebStream(stream)) {
+    throw new ERR_INVALID_ARG_TYPE('stream', ['ReadableStream', 'WritableStream', 'Stream'], stream);
+  }
+  return module.exports.addAbortSignalNoValidate(signal, stream);
+};
+
+module.exports.addAbortSignalNoValidate = function(signal, stream) {
+  if (typeof signal !== 'object' || !('aborted' in signal)) {
+    return stream;
+  }
+  const onAbort = isNodeStream(stream) ?
+    () => {
+      stream.destroy(new AbortError(undefined, { cause: signal.reason }));
+    } :
+    () => {
+      stream[kControllerErrorFunction](new AbortError(undefined, { cause: signal.reason }));
+    };
+  if (signal.aborted) {
+    onAbort();
+  } else {
+    addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
+    const disposable = addAbortListener(signal, onAbort);
+    eos(stream, disposable[SymbolDispose]);
+  }
+  return stream;
+};

lib/internal/streams/compose.js

@@ -0,0 +1,257 @@
+'use strict';
+
+const {
+  ArrayPrototypeSlice,
+} = primordials;
+
+const { pipeline } = require('internal/streams/pipeline');
+const Duplex = require('internal/streams/duplex');
+const { destroyer } = require('internal/streams/destroy');
+const {
+  isNodeStream,
+  isReadable,
+  isWritable,
+  isWebStream,
+  isTransformStream,
+  isWritableStream,
+  isReadableStream,
+} = require('internal/streams/utils');
+const {
+  AbortError,
+  codes: {
+    ERR_INVALID_ARG_VALUE,
+    ERR_MISSING_ARGS,
+  },
+} = require('internal/errors');
+const eos = require('internal/streams/end-of-stream');
+
+module.exports = function compose(...streams) {
+  if (streams.length === 0) {
+    throw new ERR_MISSING_ARGS('streams');
+  }
+
+  if (streams.length === 1) {
+    return Duplex.from(streams[0]);
+  }
+
+  const orgStreams = ArrayPrototypeSlice(streams);
+
+  if (typeof streams[0] === 'function') {
+    streams[0] = Duplex.from(streams[0]);
+  }
+
+  if (typeof streams[streams.length - 1] === 'function') {
+    const idx = streams.length - 1;
+    streams[idx] = Duplex.from(streams[idx]);
+  }
+
+  for (let n = 0; n < streams.length; ++n) {
+    if (!isNodeStream(streams[n]) && !isWebStream(streams[n])) {
+      // TODO(ronag): Add checks for non streams.
+      continue;
+    }
+    if (
+      n < streams.length - 1 &&
+      !(
+        isReadable(streams[n]) ||
+        isReadableStream(streams[n]) ||
+        isTransformStream(streams[n])
+      )
+    ) {
+      throw new ERR_INVALID_ARG_VALUE(
+        `streams[${n}]`,
+        orgStreams[n],
+        'must be readable',
+      );
+    }
+    if (
+      n > 0 &&
+      !(
+        isWritable(streams[n]) ||
+        isWritableStream(streams[n]) ||
+        isTransformStream(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);
+    } else if (!readable && !writable) {
+      d.destroy();
+    }
+  }
+
+  const head = streams[0];
+  const tail = pipeline(streams, onfinished);
+
+  const writable = !!(
+    isWritable(head) ||
+    isWritableStream(head) ||
+    isTransformStream(head)
+  );
+  const readable = !!(
+    isReadable(tail) ||
+    isReadableStream(tail) ||
+    isTransformStream(tail)
+  );
+
+  // TODO(ronag): Avoid double buffering.
+  // Implement Writable/Readable/Duplex traits.
+  // See, https://github.com/nodejs/node/pull/33515.
+  d = new Duplex({
+    // TODO (ronag): highWaterMark?
+    writableObjectMode: !!head?.writableObjectMode,
+    readableObjectMode: !!tail?.readableObjectMode,
+    writable,
+    readable,
+  });
+
+  if (writable) {
+    if (isNodeStream(head)) {
+      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();
+        }
+      });
+    } else if (isWebStream(head)) {
+      const writable = isTransformStream(head) ? head.writable : head;
+      const writer = writable.getWriter();
+
+      d._write = async function(chunk, encoding, callback) {
+        try {
+          await writer.ready;
+          writer.write(chunk).catch(() => {});
+          callback();
+        } catch (err) {
+          callback(err);
+        }
+      };
+
+      d._final = async function(callback) {
+        try {
+          await writer.ready;
+          writer.close().catch(() => {});
+          onfinish = callback;
+        } catch (err) {
+          callback(err);
+        }
+      };
+    }
+
+    const toRead = isTransformStream(tail) ? tail.readable : tail;
+
+    eos(toRead, () => {
+      if (onfinish) {
+        const cb = onfinish;
+        onfinish = null;
+        cb();
+      }
+    });
+  }
+
+  if (readable) {
+    if (isNodeStream(tail)) {
+      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;
+          }
+        }
+      };
+    } else if (isWebStream(tail)) {
+      const readable = isTransformStream(tail) ? tail.readable : tail;
+      const reader = readable.getReader();
+      d._read = async function() {
+        while (true) {
+          try {
+            const { value, done } = await reader.read();
+
+            if (!d.push(value)) {
+              return;
+            }
+
+            if (done) {
+              d.push(null);
+              return;
+            }
+          } catch {
+            return;
+          }
+        }
+      };
+    }
+  }
+
+  d._destroy = function(err, callback) {
+    if (!err && onclose !== null) {
+      err = new AbortError();
+    }
+
+    onreadable = null;
+    ondrain = null;
+    onfinish = null;
+
+    if (isNodeStream(tail)) {
+      destroyer(tail, err);
+    }
+
+    if (onclose === null) {
+      callback(err);
+    } else {
+      onclose = callback;
+    }
+  };
+
+  return d;
+};

lib/internal/streams/destroy.js

@@ -0,0 +1,361 @@
+'use strict';
+
+const {
+  Symbol,
+} = primordials;
+
+const {
+  AbortError,
+  aggregateTwoErrors,
+  codes: {
+    ERR_MULTIPLE_CALLBACK,
+  },
+} = require('internal/errors');
+const {
+  kIsDestroyed,
+  isDestroyed,
+  isFinished,
+  isServerRequest,
+  kState,
+  kErrorEmitted,
+  kEmitClose,
+  kClosed,
+  kCloseEmitted,
+  kConstructed,
+  kDestroyed,
+  kAutoDestroy,
+  kErrored,
+} = require('internal/streams/utils');
+
+const kDestroy = Symbol('kDestroy');
+const kConstruct = Symbol('kConstruct');
+
+function checkError(err, w, r) {
+  if (err) {
+    // Avoid V8 leak, https://github.com/nodejs/node/pull/34103#issuecomment-652002364
+    err.stack; // eslint-disable-line no-unused-expressions
+
+    if (w && !w.errored) {
+      w.errored = err;
+    }
+    if (r && !r.errored) {
+      r.errored = err;
+    }
+  }
+}
+
+// Backwards compat. cb() is undocumented and unused in core but
+// unfortunately might be used by modules.
+function destroy(err, cb) {
+  const r = this._readableState;
+  const w = this._writableState;
+  // With duplex streams we use the writable side for state.
+  const s = w || r;
+
+  if (
+    (w && (w[kState] & kDestroyed) !== 0) ||
+    (r && (r[kState] & kDestroyed) !== 0)
+  ) {
+    if (typeof cb === 'function') {
+      cb();
+    }
+
+    return this;
+  }
+
+
+  // We set destroyed to true before firing error callbacks in order
+  // to make it re-entrance safe in case destroy() is called within callbacks
+  checkError(err, w, r);
+
+  if (w) {
+    w[kState] |= kDestroyed;
+  }
+  if (r) {
+    r[kState] |= kDestroyed;
+  }
+
+  // If still constructing then defer calling _destroy.
+  if ((s[kState] & kConstructed) === 0) {
+    this.once(kDestroy, function(er) {
+      _destroy(this, aggregateTwoErrors(er, err), cb);
+    });
+  } else {
+    _destroy(this, err, cb);
+  }
+
+  return this;
+}
+
+function _destroy(self, err, cb) {
+  let called = false;
+
+  function onDestroy(err) {
+    if (called) {
+      return;
+    }
+    called = true;
+
+    const r = self._readableState;
+    const w = self._writableState;
+
+    checkError(err, w, r);
+
+    if (w) {
+      w[kState] |= kClosed;
+    }
+    if (r) {
+      r[kState] |= kClosed;
+    }
+
+    if (typeof cb === 'function') {
+      cb(err);
+    }
+
+    if (err) {
+      process.nextTick(emitErrorCloseNT, self, err);
+    } else {
+      process.nextTick(emitCloseNT, self);
+    }
+  }
+  try {
+    self._destroy(err || null, onDestroy);
+  } catch (err) {
+    onDestroy(err);
+  }
+}
+
+function emitErrorCloseNT(self, err) {
+  emitErrorNT(self, err);
+  emitCloseNT(self);
+}
+
+function emitCloseNT(self) {
+  const r = self._readableState;
+  const w = self._writableState;
+
+  if (w) {
+    w[kState] |= kCloseEmitted;
+  }
+  if (r) {
+    r[kState] |= kCloseEmitted;
+  }
+
+  if (
+    (w && (w[kState] & kEmitClose) !== 0) ||
+    (r && (r[kState] & kEmitClose) !== 0)
+  ) {
+    self.emit('close');
+  }
+}
+
+function emitErrorNT(self, err) {
+  const r = self._readableState;
+  const w = self._writableState;
+
+  if (
+    (w && (w[kState] & kErrorEmitted) !== 0) ||
+    (r && (r[kState] & kErrorEmitted) !== 0)
+  ) {
+    return;
+  }
+
+  if (w) {
+    w[kState] |= kErrorEmitted;
+  }
+  if (r) {
+    r[kState] |= kErrorEmitted;
+  }
+
+  self.emit('error', err);
+}
+
+function undestroy() {
+  const r = this._readableState;
+  const w = this._writableState;
+
+  if (r) {
+    r.constructed = true;
+    r.closed = false;
+    r.closeEmitted = false;
+    r.destroyed = false;
+    r.errored = null;
+    r.errorEmitted = false;
+    r.reading = false;
+    r.ended = r.readable === false;
+    r.endEmitted = r.readable === false;
+  }
+
+  if (w) {
+    w.constructed = true;
+    w.destroyed = false;
+    w.closed = false;
+    w.closeEmitted = false;
+    w.errored = null;
+    w.errorEmitted = false;
+    w.finalCalled = false;
+    w.prefinished = false;
+    w.ended = w.writable === false;
+    w.ending = w.writable === false;
+    w.finished = w.writable === false;
+  }
+}
+
+function errorOrDestroy(stream, err, sync) {
+  // We have tests that rely on errors being emitted
+  // in the same tick, so changing this is semver major.
+  // For now when you opt-in to autoDestroy we allow
+  // the error to be emitted nextTick. In a future
+  // semver major update we should change the default to this.
+
+  const r = stream._readableState;
+  const w = stream._writableState;
+
+  if (
+    (w && (w[kState] ? (w[kState] & kDestroyed) !== 0 : w.destroyed)) ||
+    (r && (r[kState] ? (r[kState] & kDestroyed) !== 0 : r.destroyed))
+  ) {
+    return this;
+  }
+
+  if (
+    (r && (r[kState] & kAutoDestroy) !== 0) ||
+    (w && (w[kState] & kAutoDestroy) !== 0)
+  ) {
+    stream.destroy(err);
+  } else if (err) {
+    // Avoid V8 leak, https://github.com/nodejs/node/pull/34103#issuecomment-652002364
+    err.stack; // eslint-disable-line no-unused-expressions
+
+    if (w && (w[kState] & kErrored) === 0) {
+      w.errored = err;
+    }
+    if (r && (r[kState] & kErrored) === 0) {
+      r.errored = err;
+    }
+    if (sync) {
+      process.nextTick(emitErrorNT, stream, err);
+    } else {
+      emitErrorNT(stream, err);
+    }
+  }
+}
+
+function construct(stream, cb) {
+  if (typeof stream._construct !== 'function') {
+    return;
+  }
+
+  const r = stream._readableState;
+  const w = stream._writableState;
+
+  if (r) {
+    r[kState] &= ~kConstructed;
+  }
+  if (w) {
+    w[kState] &= ~kConstructed;
+  }
+
+  stream.once(kConstruct, cb);
+
+  if (stream.listenerCount(kConstruct) > 1) {
+    // Duplex
+    return;
+  }
+
+  process.nextTick(constructNT, stream);
+}
+
+function constructNT(stream) {
+  let called = false;
+
+  function onConstruct(err) {
+    if (called) {
+      errorOrDestroy(stream, err ?? new ERR_MULTIPLE_CALLBACK());
+      return;
+    }
+    called = true;
+
+    const r = stream._readableState;
+    const w = stream._writableState;
+    const s = w || r;
+
+    if (r) {
+      r[kState] |= kConstructed;
+    }
+    if (w) {
+      w[kState] |= kConstructed;
+    }
+
+    if (s.destroyed) {
+      stream.emit(kDestroy, err);
+    } else if (err) {
+      errorOrDestroy(stream, err, true);
+    } else {
+      stream.emit(kConstruct);
+    }
+  }
+
+  try {
+    stream._construct((err) => {
+      process.nextTick(onConstruct, err);
+    });
+  } catch (err) {
+    process.nextTick(onConstruct, err);
+  }
+}
+
+function isRequest(stream) {
+  return stream?.setHeader && typeof stream.abort === 'function';
+}
+
+function emitCloseLegacy(stream) {
+  stream.emit('close');
+}
+
+function emitErrorCloseLegacy(stream, err) {
+  stream.emit('error', err);
+  process.nextTick(emitCloseLegacy, stream);
+}
+
+// Normalize destroy for legacy.
+function destroyer(stream, err) {
+  if (!stream || isDestroyed(stream)) {
+    return;
+  }
+
+  if (!err && !isFinished(stream)) {
+    err = new AbortError();
+  }
+
+  // TODO: Remove isRequest branches.
+  if (isServerRequest(stream)) {
+    stream.socket = null;
+    stream.destroy(err);
+  } else if (isRequest(stream)) {
+    stream.abort();
+  } else if (isRequest(stream.req)) {
+    stream.req.abort();
+  } else if (typeof stream.destroy === 'function') {
+    stream.destroy(err);
+  } else if (typeof stream.close === 'function') {
+    // TODO: Don't lose err?
+    stream.close();
+  } else if (err) {
+    process.nextTick(emitErrorCloseLegacy, stream, err);
+  } else {
+    process.nextTick(emitCloseLegacy, stream);
+  }
+
+  if (!stream.destroyed) {
+    stream[kIsDestroyed] = true;
+  }
+}
+
+module.exports = {
+  construct,
+  destroyer,
+  destroy,
+  undestroy,
+  errorOrDestroy,
+};

lib/internal/streams/duplex.js

@@ -0,0 +1,203 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+// a duplex stream is just a stream that is both readable and writable.
+// Since JS doesn't have multiple prototype inheritance, this class
+// prototypically inherits from Readable, and then parasitically from
+// Writable.
+
+'use strict';
+
+const {
+  ObjectDefineProperties,
+  ObjectGetOwnPropertyDescriptor,
+  ObjectKeys,
+  ObjectSetPrototypeOf,
+} = primordials;
+
+module.exports = Duplex;
+
+const Stream = require('internal/streams/legacy').Stream;
+const Readable = require('internal/streams/readable');
+const Writable = require('internal/streams/writable');
+
+const {
+  addAbortSignal,
+} = require('internal/streams/add-abort-signal');
+
+const destroyImpl = require('internal/streams/destroy');
+const { kOnConstructed } = require('internal/streams/utils');
+
+ObjectSetPrototypeOf(Duplex.prototype, Readable.prototype);
+ObjectSetPrototypeOf(Duplex, Readable);
+
+{
+  const keys = ObjectKeys(Writable.prototype);
+  // Allow the keys array to be GC'ed.
+  for (let i = 0; i < keys.length; i++) {
+    const method = keys[i];
+    Duplex.prototype[method] ||= Writable.prototype[method];
+  }
+}
+
+// Use the `destroy` method of `Writable`.
+Duplex.prototype.destroy = Writable.prototype.destroy;
+
+function Duplex(options) {
+  if (!(this instanceof Duplex))
+    return new Duplex(options);
+
+  this._events ??= {
+    close: undefined,
+    error: undefined,
+    prefinish: undefined,
+    finish: undefined,
+    drain: undefined,
+    data: undefined,
+    end: undefined,
+    readable: undefined,
+    // Skip uncommon events...
+    // pause: undefined,
+    // resume: undefined,
+    // pipe: undefined,
+    // unpipe: undefined,
+    // [destroyImpl.kConstruct]: undefined,
+    // [destroyImpl.kDestroy]: undefined,
+  };
+
+  this._readableState = new Readable.ReadableState(options, this, true);
+  this._writableState = new Writable.WritableState(options, this, true);
+
+  if (options) {
+    this.allowHalfOpen = options.allowHalfOpen !== false;
+
+    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;
+    }
+
+    if (typeof options.read === 'function')
+      this._read = options.read;
+
+    if (typeof options.write === 'function')
+      this._write = options.write;
+
+    if (typeof options.writev === 'function')
+      this._writev = options.writev;
+
+    if (typeof options.destroy === 'function')
+      this._destroy = options.destroy;
+
+    if (typeof options.final === 'function')
+      this._final = options.final;
+
+    if (typeof options.construct === 'function')
+      this._construct = options.construct;
+
+    if (options.signal)
+      addAbortSignal(options.signal, this);
+  } else {
+    this.allowHalfOpen = true;
+  }
+
+  Stream.call(this, options);
+
+  if (this._construct != null) {
+    destroyImpl.construct(this, () => {
+      this._readableState[kOnConstructed](this);
+      this._writableState[kOnConstructed](this);
+    });
+  }
+}
+
+ObjectDefineProperties(Duplex.prototype, {
+  writable:
+    { __proto__: null, ...ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writable') },
+  writableHighWaterMark:
+    { __proto__: null, ...ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableHighWaterMark') },
+  writableObjectMode:
+    { __proto__: null, ...ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableObjectMode') },
+  writableBuffer:
+    { __proto__: null, ...ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableBuffer') },
+  writableLength:
+    { __proto__: null, ...ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableLength') },
+  writableFinished:
+    { __proto__: null, ...ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableFinished') },
+  writableCorked:
+    { __proto__: null, ...ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableCorked') },
+  writableEnded:
+    { __proto__: null, ...ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableEnded') },
+  writableNeedDrain:
+    { __proto__: null, ...ObjectGetOwnPropertyDescriptor(Writable.prototype, 'writableNeedDrain') },
+
+  destroyed: {
+    __proto__: null,
+    get() {
+      if (this._readableState === undefined ||
+        this._writableState === undefined) {
+        return false;
+      }
+      return this._readableState.destroyed && this._writableState.destroyed;
+    },
+    set(value) {
+      // Backward compatibility, the user is explicitly
+      // managing destroyed.
+      if (this._readableState && this._writableState) {
+        this._readableState.destroyed = value;
+        this._writableState.destroyed = value;
+      }
+    },
+  },
+});
+
+let webStreamsAdapters;
+
+// Lazy to avoid circular references
+function lazyWebStreams() {
+  if (webStreamsAdapters === undefined)
+    webStreamsAdapters = require('internal/webstreams/adapters');
+  return webStreamsAdapters;
+}
+
+Duplex.fromWeb = function(pair, options) {
+  return lazyWebStreams().newStreamDuplexFromReadableWritablePair(
+    pair,
+    options);
+};
+
+Duplex.toWeb = function(duplex) {
+  return lazyWebStreams().newReadableWritablePairFromDuplex(duplex);
+};
+
+let duplexify;
+
+Duplex.from = function(body) {
+  duplexify ??= require('internal/streams/duplexify');
+  return duplexify(body, 'body');
+};

lib/internal/streams/duplexify.js

@@ -0,0 +1,379 @@
+'use strict';
+
+const {
+  FunctionPrototypeCall,
+  PromiseWithResolvers,
+} = primordials;
+
+const {
+  isReadable,
+  isWritable,
+  isIterable,
+  isNodeStream,
+  isReadableNodeStream,
+  isWritableNodeStream,
+  isDuplexNodeStream,
+  isReadableStream,
+  isWritableStream,
+} = 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 from = require('internal/streams/from');
+
+const {
+  isBlob,
+} = require('internal/blob');
+const { AbortController } = require('internal/abort_controller');
+
+// 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 });
+  }
+
+  if (isReadableStream(body)) {
+    return _duplexify({ readable: Readable.fromWeb(body) });
+  }
+
+  if (isWritableStream(body)) {
+    return _duplexify({ writable: Writable.fromWeb(body) });
+  }
+
+  if (typeof body === 'function') {
+    const { value, write, final, destroy } = fromAsyncGen(body);
+
+    // Body might be a constructor function instead of an async generator function.
+    if (isDuplexNodeStream(value)) {
+      return value;
+    }
+
+    if (isIterable(value)) {
+      return from(Duplexify, value, {
+        // TODO (ronag): highWaterMark?
+        objectMode: true,
+        write,
+        final,
+        destroy,
+      });
+    }
+
+    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);
+            }
+          });
+        },
+        destroy,
+      });
+    }
+
+    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,
+    });
+  }
+
+  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 } = PromiseWithResolvers();
+  const ac = new AbortController();
+  const signal = ac.signal;
+  const value = fn(async function*() {
+    while (true) {
+      const _promise = promise;
+      promise = null;
+      const { chunk, done, cb } = await _promise;
+      process.nextTick(cb);
+      if (done) return;
+      if (signal.aborted)
+        throw new AbortError(undefined, { cause: signal.reason });
+      ({ promise, resolve } = PromiseWithResolvers());
+      yield chunk;
+    }
+  }(), { signal });
+
+  return {
+    value,
+    write(chunk, encoding, cb) {
+      const _resolve = resolve;
+      resolve = null;
+      _resolve({ chunk, done: false, cb });
+    },
+    final(cb) {
+      const _resolve = resolve;
+      resolve = null;
+      _resolve({ done: true, cb });
+    },
+    destroy(err, cb) {
+      ac.abort();
+      cb(err);
+    },
+  };
+}
+
+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);
+    }
+  }
+
+  // TODO(ronag): Avoid double buffering.
+  // Implement Writable/Readable/Duplex traits.
+  // See, https://github.com/nodejs/node/pull/33515.
+  d = new Duplexify({
+    // TODO (ronag): highWaterMark?
+    readableObjectMode: !!r?.readableObjectMode,
+    writableObjectMode: !!w?.writableObjectMode,
+    readable,
+    writable,
+  });
+
+  if (writable) {
+    eos(w, (err) => {
+      writable = false;
+      if (err) {
+        destroyer(r, err);
+      }
+      onfinished(err);
+    });
+
+    d._write = function(chunk, encoding, callback) {
+      if (w.write(chunk, encoding)) {
+        callback();
+      } else {
+        ondrain = callback;
+      }
+    };
+
+    d._final = function(callback) {
+      w.end();
+      onfinish = callback;
+    };
+
+    w.on('drain', function() {
+      if (ondrain) {
+        const cb = ondrain;
+        ondrain = null;
+        cb();
+      }
+    });
+
+    w.on('finish', function() {
+      if (onfinish) {
+        const cb = onfinish;
+        onfinish = null;
+        cb();
+      }
+    });
+  }
+
+  if (readable) {
+    eos(r, (err) => {
+      readable = false;
+      if (err) {
+        destroyer(r, err);
+      }
+      onfinished(err);
+    });
+
+    r.on('readable', function() {
+      if (onreadable) {
+        const cb = onreadable;
+        onreadable = null;
+        cb();
+      }
+    });
+
+    r.on('end', function() {
+      d.push(null);
+    });
+
+    d._read = function() {
+      while (true) {
+        const buf = r.read();
+
+        if (buf === null) {
+          onreadable = d._read;
+          return;
+        }
+
+        if (!d.push(buf)) {
+          return;
+        }
+      }
+    };
+  }
+
+  d._destroy = function(err, callback) {
+    if (!err && onclose !== null) {
+      err = new AbortError();
+    }
+
+    onreadable = null;
+    ondrain = null;
+    onfinish = null;
+
+    if (onclose === null) {
+      callback(err);
+    } else {
+      onclose = callback;
+      destroyer(w, err);
+      destroyer(r, err);
+    }
+  };
+
+  return d;
+}

lib/internal/streams/duplexpair.js

@@ -0,0 +1,62 @@
+'use strict';
+const {
+  Symbol,
+} = primordials;
+
+const { Duplex } = require('stream');
+const assert = require('internal/assert');
+
+const kCallback = Symbol('Callback');
+const kInitOtherSide = Symbol('InitOtherSide');
+
+class DuplexSide extends Duplex {
+  #otherSide = null;
+
+  constructor(options) {
+    super(options);
+    this[kCallback] = null;
+    this.#otherSide = null;
+  }
+
+  [kInitOtherSide](otherSide) {
+    // Ensure this can only be set once, to enforce encapsulation.
+    if (this.#otherSide === null) {
+      this.#otherSide = otherSide;
+    } else {
+      assert(this.#otherSide === null);
+    }
+  }
+
+  _read() {
+    const callback = this[kCallback];
+    if (callback) {
+      this[kCallback] = null;
+      callback();
+    }
+  }
+
+  _write(chunk, encoding, callback) {
+    assert(this.#otherSide !== null);
+    assert(this.#otherSide[kCallback] === null);
+    if (chunk.length === 0) {
+      process.nextTick(callback);
+    } else {
+      this.#otherSide.push(chunk);
+      this.#otherSide[kCallback] = callback;
+    }
+  }
+
+  _final(callback) {
+    this.#otherSide.on('end', callback);
+    this.#otherSide.push(null);
+  }
+}
+
+function duplexPair(options) {
+  const side0 = new DuplexSide(options);
+  const side1 = new DuplexSide(options);
+  side0[kInitOtherSide](side1);
+  side1[kInitOtherSide](side0);
+  return [ side0, side1 ];
+}
+module.exports = duplexPair;

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

@@ -0,0 +1,331 @@
+// Ported from https://github.com/mafintosh/end-of-stream with
+// permission from the author, Mathias Buus (@mafintosh).
+
+'use strict';
+
+const {
+  Promise,
+  PromisePrototypeThen,
+  SymbolDispose,
+} = primordials;
+
+const {
+  AbortError,
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+    ERR_STREAM_PREMATURE_CLOSE,
+  },
+} = require('internal/errors');
+const {
+  kEmptyObject,
+  once,
+} = require('internal/util');
+const {
+  validateAbortSignal,
+  validateFunction,
+  validateObject,
+  validateBoolean,
+} = require('internal/validators');
+
+const {
+  isClosed,
+  isReadable,
+  isReadableNodeStream,
+  isReadableStream,
+  isReadableFinished,
+  isReadableErrored,
+  isWritable,
+  isWritableNodeStream,
+  isWritableStream,
+  isWritableFinished,
+  isWritableErrored,
+  isNodeStream,
+  willEmitClose: _willEmitClose,
+  kIsClosedPromise,
+} = require('internal/streams/utils');
+
+// Lazy load
+let AsyncLocalStorage;
+let addAbortListener;
+
+function isRequest(stream) {
+  return stream.setHeader && typeof stream.abort === 'function';
+}
+
+const nop = () => {};
+
+function eos(stream, options, callback) {
+  if (arguments.length === 2) {
+    callback = options;
+    options = kEmptyObject;
+  } else if (options == null) {
+    options = kEmptyObject;
+  } else {
+    validateObject(options, 'options');
+  }
+  validateFunction(callback, 'callback');
+  validateAbortSignal(options.signal, 'options.signal');
+
+  AsyncLocalStorage ??= require('async_hooks').AsyncLocalStorage;
+  callback = once(AsyncLocalStorage.bind(callback));
+
+  if (isReadableStream(stream) || isWritableStream(stream)) {
+    return eosWeb(stream, options, callback);
+  }
+
+  if (!isNodeStream(stream)) {
+    throw new ERR_INVALID_ARG_TYPE('stream', ['ReadableStream', 'WritableStream', 'Stream'], stream);
+  }
+
+  const readable = options.readable ?? isReadableNodeStream(stream);
+  const writable = options.writable ?? isWritableNodeStream(stream);
+
+  const wState = stream._writableState;
+  const rState = stream._readableState;
+
+  const onlegacyfinish = () => {
+    if (!stream.writable) {
+      onfinish();
+    }
+  };
+
+  // TODO (ronag): Improve soft detection to include core modules and
+  // common ecosystem modules that do properly emit 'close' but fail
+  // this generic check.
+  let willEmitClose = (
+    _willEmitClose(stream) &&
+    isReadableNodeStream(stream) === readable &&
+    isWritableNodeStream(stream) === writable
+  );
+
+  let writableFinished = isWritableFinished(stream, false);
+  const onfinish = () => {
+    writableFinished = true;
+    // Stream should not be destroyed here. If it is that
+    // means that user space is doing something differently and
+    // we cannot trust willEmitClose.
+    if (stream.destroyed) {
+      willEmitClose = false;
+    }
+
+    if (willEmitClose && (!stream.readable || readable)) {
+      return;
+    }
+
+    if (!readable || readableFinished) {
+      callback.call(stream);
+    }
+  };
+
+  let readableFinished = isReadableFinished(stream, false);
+  const onend = () => {
+    readableFinished = true;
+    // Stream should not be destroyed here. If it is that
+    // means that user space is doing something differently and
+    // we cannot trust willEmitClose.
+    if (stream.destroyed) {
+      willEmitClose = false;
+    }
+
+    if (willEmitClose && (!stream.writable || writable)) {
+      return;
+    }
+
+    if (!writable || writableFinished) {
+      callback.call(stream);
+    }
+  };
+
+  const onerror = (err) => {
+    callback.call(stream, err);
+  };
+
+  let closed = isClosed(stream);
+
+  const onclose = () => {
+    closed = true;
+
+    const errored = isWritableErrored(stream) || isReadableErrored(stream);
+
+    if (errored && typeof errored !== 'boolean') {
+      return callback.call(stream, errored);
+    }
+
+    if (readable && !readableFinished && isReadableNodeStream(stream, true)) {
+      if (!isReadableFinished(stream, false))
+        return callback.call(stream,
+                             new ERR_STREAM_PREMATURE_CLOSE());
+    }
+    if (writable && !writableFinished) {
+      if (!isWritableFinished(stream, false))
+        return callback.call(stream,
+                             new ERR_STREAM_PREMATURE_CLOSE());
+    }
+
+    callback.call(stream);
+  };
+
+  const onclosed = () => {
+    closed = true;
+
+    const errored = isWritableErrored(stream) || isReadableErrored(stream);
+
+    if (errored && typeof errored !== 'boolean') {
+      return callback.call(stream, errored);
+    }
+
+    callback.call(stream);
+  };
+
+  const onrequest = () => {
+    stream.req.on('finish', onfinish);
+  };
+
+  if (isRequest(stream)) {
+    stream.on('complete', onfinish);
+    if (!willEmitClose) {
+      stream.on('abort', onclose);
+    }
+    if (stream.req) {
+      onrequest();
+    } else {
+      stream.on('request', onrequest);
+    }
+  } else if (writable && !wState) { // legacy streams
+    stream.on('end', onlegacyfinish);
+    stream.on('close', onlegacyfinish);
+  }
+
+  // Not all streams will emit 'close' after 'aborted'.
+  if (!willEmitClose && typeof stream.aborted === 'boolean') {
+    stream.on('aborted', onclose);
+  }
+
+  stream.on('end', onend);
+  stream.on('finish', onfinish);
+  if (options.error !== false) {
+    stream.on('error', onerror);
+  }
+  stream.on('close', onclose);
+
+  if (closed) {
+    process.nextTick(onclose);
+  } else if (wState?.errorEmitted || rState?.errorEmitted) {
+    if (!willEmitClose) {
+      process.nextTick(onclosed);
+    }
+  } else if (
+    !readable &&
+    (!willEmitClose || isReadable(stream)) &&
+    (writableFinished || isWritable(stream) === false) &&
+    (wState == null || wState.pendingcb === undefined || wState.pendingcb === 0)
+  ) {
+    process.nextTick(onclosed);
+  } else if (
+    !writable &&
+    (!willEmitClose || isWritable(stream)) &&
+    (readableFinished || isReadable(stream) === false)
+  ) {
+    process.nextTick(onclosed);
+  } else if ((rState && stream.req && stream.aborted)) {
+    process.nextTick(onclosed);
+  }
+
+  const cleanup = () => {
+    callback = nop;
+    stream.removeListener('aborted', onclose);
+    stream.removeListener('complete', onfinish);
+    stream.removeListener('abort', onclose);
+    stream.removeListener('request', onrequest);
+    if (stream.req) stream.req.removeListener('finish', onfinish);
+    stream.removeListener('end', onlegacyfinish);
+    stream.removeListener('close', onlegacyfinish);
+    stream.removeListener('finish', onfinish);
+    stream.removeListener('end', onend);
+    stream.removeListener('error', onerror);
+    stream.removeListener('close', onclose);
+  };
+
+  if (options.signal && !closed) {
+    const abort = () => {
+      // Keep it because cleanup removes it.
+      const endCallback = callback;
+      cleanup();
+      endCallback.call(
+        stream,
+        new AbortError(undefined, { cause: options.signal.reason }));
+    };
+    if (options.signal.aborted) {
+      process.nextTick(abort);
+    } else {
+      addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
+      const disposable = addAbortListener(options.signal, abort);
+      const originalCallback = callback;
+      callback = once((...args) => {
+        disposable[SymbolDispose]();
+        originalCallback.apply(stream, args);
+      });
+    }
+  }
+
+  return cleanup;
+}
+
+function eosWeb(stream, options, callback) {
+  let isAborted = false;
+  let abort = nop;
+  if (options.signal) {
+    abort = () => {
+      isAborted = true;
+      callback.call(stream, new AbortError(undefined, { cause: options.signal.reason }));
+    };
+    if (options.signal.aborted) {
+      process.nextTick(abort);
+    } else {
+      addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
+      const disposable = addAbortListener(options.signal, abort);
+      const originalCallback = callback;
+      callback = once((...args) => {
+        disposable[SymbolDispose]();
+        originalCallback.apply(stream, args);
+      });
+    }
+  }
+  const resolverFn = (...args) => {
+    if (!isAborted) {
+      process.nextTick(() => callback.apply(stream, args));
+    }
+  };
+  PromisePrototypeThen(
+    stream[kIsClosedPromise].promise,
+    resolverFn,
+    resolverFn,
+  );
+  return nop;
+}
+
+function finished(stream, opts) {
+  let autoCleanup = false;
+  if (opts === null) {
+    opts = kEmptyObject;
+  }
+  if (opts?.cleanup) {
+    validateBoolean(opts.cleanup, 'cleanup');
+    autoCleanup = opts.cleanup;
+  }
+  return new Promise((resolve, reject) => {
+    const cleanup = eos(stream, opts, (err) => {
+      if (autoCleanup) {
+        cleanup();
+      }
+      if (err) {
+        reject(err);
+      } else {
+        resolve();
+      }
+    });
+  });
+}
+
+module.exports = eos;
+module.exports.finished = finished;

lib/internal/streams/from.js

@@ -0,0 +1,208 @@
+'use strict';
+
+const {
+  PromisePrototypeThen,
+  SymbolAsyncIterator,
+  SymbolIterator,
+} = primordials;
+const { Buffer } = require('buffer');
+
+const {
+  ERR_INVALID_ARG_TYPE,
+  ERR_STREAM_NULL_VALUES,
+} = require('internal/errors').codes;
+
+function from(Readable, iterable, opts) {
+  let iterator;
+  if (typeof iterable === 'string' || iterable instanceof Buffer) {
+    return new Readable({
+      objectMode: true,
+      ...opts,
+      read() {
+        this.push(iterable);
+        this.push(null);
+      },
+    });
+  }
+
+  let isAsync;
+  if (iterable?.[SymbolAsyncIterator]) {
+    isAsync = true;
+    iterator = iterable[SymbolAsyncIterator]();
+  } else if (iterable?.[SymbolIterator]) {
+    isAsync = false;
+    iterator = iterable[SymbolIterator]();
+  } else {
+    throw new ERR_INVALID_ARG_TYPE('iterable', ['Iterable'], iterable);
+  }
+
+
+  const readable = new Readable({
+    objectMode: true,
+    highWaterMark: 1,
+    // TODO(ronag): What options should be allowed?
+    ...opts,
+  });
+
+  // Flag to protect against _read
+  // being called before last iteration completion.
+  let reading = false;
+  let isAsyncValues = false;
+
+  readable._read = function() {
+    if (!reading) {
+      reading = true;
+
+      if (isAsync) {
+        nextAsync();
+      } else if (isAsyncValues) {
+        nextSyncWithAsyncValues();
+      } else {
+        nextSyncWithSyncValues();
+      }
+    }
+  };
+
+  readable._destroy = function(error, cb) {
+    PromisePrototypeThen(
+      close(error),
+      () => process.nextTick(cb, error), // nextTick is here in case cb throws
+      (e) => process.nextTick(cb, e || error),
+    );
+  };
+
+  async function close(error) {
+    const hadError = (error !== undefined) && (error !== null);
+    const hasThrow = typeof iterator.throw === 'function';
+    if (hadError && hasThrow) {
+      const { value, done } = await iterator.throw(error);
+      await value;
+      if (done) {
+        return;
+      }
+    }
+    if (typeof iterator.return === 'function') {
+      const { value } = await iterator.return();
+      await value;
+    }
+  }
+
+  // There are a lot of duplication here, it's done on purpose for performance
+  // reasons - avoid await when not needed.
+
+  function nextSyncWithSyncValues() {
+    for (;;) {
+      try {
+        const { value, done } = iterator.next();
+
+        if (done) {
+          readable.push(null);
+          return;
+        }
+
+        if (value &&
+          typeof value.then === 'function') {
+          return changeToAsyncValues(value);
+        }
+
+        if (value === null) {
+          reading = false;
+          throw new ERR_STREAM_NULL_VALUES();
+        }
+
+        if (readable.push(value)) {
+          continue;
+        }
+
+        reading = false;
+      } catch (err) {
+        readable.destroy(err);
+      }
+      break;
+    }
+  }
+
+  async function changeToAsyncValues(value) {
+    isAsyncValues = true;
+
+    try {
+      const res = await value;
+
+      if (res === null) {
+        reading = false;
+        throw new ERR_STREAM_NULL_VALUES();
+      }
+
+      if (readable.push(res)) {
+        nextSyncWithAsyncValues();
+        return;
+      }
+
+      reading = false;
+    } catch (err) {
+      readable.destroy(err);
+    }
+  }
+
+  async function nextSyncWithAsyncValues() {
+    for (;;) {
+      try {
+        const { value, done } = iterator.next();
+
+        if (done) {
+          readable.push(null);
+          return;
+        }
+
+        const res = (value &&
+          typeof value.then === 'function') ?
+          await value :
+          value;
+
+        if (res === null) {
+          reading = false;
+          throw new ERR_STREAM_NULL_VALUES();
+        }
+
+        if (readable.push(res)) {
+          continue;
+        }
+
+        reading = false;
+      } catch (err) {
+        readable.destroy(err);
+      }
+      break;
+    }
+  }
+
+  async function nextAsync() {
+    for (;;) {
+      try {
+        const { value, done } = await iterator.next();
+
+        if (done) {
+          readable.push(null);
+          return;
+        }
+
+        if (value === null) {
+          reading = false;
+          throw new ERR_STREAM_NULL_VALUES();
+        }
+
+        if (readable.push(value)) {
+          continue;
+        }
+
+        reading = false;
+      } catch (err) {
+        readable.destroy(err);
+      }
+      break;
+    }
+  }
+  return readable;
+}
+
+module.exports = from;

lib/internal/streams/lazy_transform.js

@@ -0,0 +1,57 @@
+// LazyTransform is a special type of Transform stream that is lazily loaded.
+// This is used for performance with bi-API-ship: when two APIs are available
+// for the stream, one conventional and one non-conventional.
+'use strict';
+
+const {
+  ObjectDefineProperties,
+  ObjectDefineProperty,
+  ObjectSetPrototypeOf,
+} = primordials;
+
+const stream = require('stream');
+
+module.exports = LazyTransform;
+
+function LazyTransform(options) {
+  this._options = options;
+}
+ObjectSetPrototypeOf(LazyTransform.prototype, stream.Transform.prototype);
+ObjectSetPrototypeOf(LazyTransform, stream.Transform);
+
+function makeGetter(name) {
+  return function() {
+    stream.Transform.call(this, this._options);
+    this._writableState.decodeStrings = false;
+    return this[name];
+  };
+}
+
+function makeSetter(name) {
+  return function(val) {
+    ObjectDefineProperty(this, name, {
+      __proto__: null,
+      value: val,
+      enumerable: true,
+      configurable: true,
+      writable: true,
+    });
+  };
+}
+
+ObjectDefineProperties(LazyTransform.prototype, {
+  _readableState: {
+    __proto__: null,
+    get: makeGetter('_readableState'),
+    set: makeSetter('_readableState'),
+    configurable: true,
+    enumerable: true,
+  },
+  _writableState: {
+    __proto__: null,
+    get: makeGetter('_writableState'),
+    set: makeSetter('_writableState'),
+    configurable: true,
+    enumerable: true,
+  },
+});

lib/internal/streams/legacy.js

@@ -0,0 +1,125 @@
+'use strict';
+
+const {
+  ArrayIsArray,
+  ObjectSetPrototypeOf,
+  ReflectOwnKeys,
+} = primordials;
+
+const EE = require('events');
+
+function Stream(opts) {
+  EE.call(this, opts);
+}
+ObjectSetPrototypeOf(Stream.prototype, EE.prototype);
+ObjectSetPrototypeOf(Stream, EE);
+
+Stream.prototype.pipe = function(dest, options) {
+  const source = this;
+
+  function ondata(chunk) {
+    if (dest.writable && dest.write(chunk) === false && source.pause) {
+      source.pause();
+    }
+  }
+
+  source.on('data', ondata);
+
+  function ondrain() {
+    if (source.readable && source.resume) {
+      source.resume();
+    }
+  }
+
+  dest.on('drain', ondrain);
+
+  // If the 'end' option is not supplied, dest.end() will be called when
+  // source gets the 'end' or 'close' events.  Only dest.end() once.
+  if (!dest._isStdio && (!options || options.end !== false)) {
+    source.on('end', onend);
+    source.on('close', onclose);
+  }
+
+  let didOnEnd = false;
+  function onend() {
+    if (didOnEnd) return;
+    didOnEnd = true;
+
+    dest.end();
+  }
+
+
+  function onclose() {
+    if (didOnEnd) return;
+    didOnEnd = true;
+
+    if (typeof dest.destroy === 'function') dest.destroy();
+  }
+
+  // Don't leave dangling pipes when there are errors.
+  function onerror(er) {
+    cleanup();
+    if (EE.listenerCount(this, 'error') === 0) {
+      this.emit('error', er);
+    }
+  }
+
+  prependListener(source, 'error', onerror);
+  prependListener(dest, 'error', onerror);
+
+  // Remove all the event listeners that were added.
+  function cleanup() {
+    source.removeListener('data', ondata);
+    dest.removeListener('drain', ondrain);
+
+    source.removeListener('end', onend);
+    source.removeListener('close', onclose);
+
+    source.removeListener('error', onerror);
+    dest.removeListener('error', onerror);
+
+    source.removeListener('end', cleanup);
+    source.removeListener('close', cleanup);
+
+    dest.removeListener('close', cleanup);
+  }
+
+  source.on('end', cleanup);
+  source.on('close', cleanup);
+
+  dest.on('close', cleanup);
+  dest.emit('pipe', source);
+
+  // Allow for unix-like usage: A.pipe(B).pipe(C)
+  return dest;
+};
+
+Stream.prototype.eventNames = function eventNames() {
+  const names = [];
+  for (const key of ReflectOwnKeys(this._events)) {
+    if (typeof this._events[key] === 'function' || (ArrayIsArray(this._events[key]) && this._events[key].length > 0)) {
+      names.push(key);
+    }
+  }
+  return names;
+};
+
+function prependListener(emitter, event, fn) {
+  // Sadly this is not cacheable as some libraries bundle their own
+  // event emitter implementation with them.
+  if (typeof emitter.prependListener === 'function')
+    return emitter.prependListener(event, fn);
+
+  // This is a hack to make sure that our error handler is attached before any
+  // userland ones.  NEVER DO THIS. This is here only because this code needs
+  // to continue to work with older versions of Node.js that do not include
+  // the prependListener() method. The goal is to eventually remove this hack.
+  if (!emitter._events || !emitter._events[event])
+    emitter.on(event, fn);
+  else if (ArrayIsArray(emitter._events[event]))
+    emitter._events[event].unshift(fn);
+  else
+    emitter._events[event] = [fn, emitter._events[event]];
+}
+
+module.exports = { Stream, prependListener };

lib/internal/streams/operators.js

@@ -0,0 +1,436 @@
+'use strict';
+
+const {
+  ArrayPrototypePush,
+  Boolean,
+  MathFloor,
+  Number,
+  NumberIsNaN,
+  Promise,
+  PromisePrototypeThen,
+  PromiseReject,
+  PromiseResolve,
+  Symbol,
+} = primordials;
+
+const { AbortController, AbortSignal } = require('internal/abort_controller');
+
+const {
+  AbortError,
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+    ERR_INVALID_ARG_VALUE,
+    ERR_MISSING_ARGS,
+    ERR_OUT_OF_RANGE,
+  },
+} = require('internal/errors');
+const {
+  validateAbortSignal,
+  validateInteger,
+  validateObject,
+} = require('internal/validators');
+const { kWeakHandler, kResistStopPropagation } = require('internal/event_target');
+const { finished } = require('internal/streams/end-of-stream');
+const staticCompose = require('internal/streams/compose');
+const {
+  addAbortSignalNoValidate,
+} = require('internal/streams/add-abort-signal');
+const { isWritable, isNodeStream } = require('internal/streams/utils');
+
+const kEmpty = Symbol('kEmpty');
+const kEof = Symbol('kEof');
+
+function compose(stream, options) {
+  if (options != null) {
+    validateObject(options, 'options');
+  }
+  if (options?.signal != null) {
+    validateAbortSignal(options.signal, 'options.signal');
+  }
+
+  if (isNodeStream(stream) && !isWritable(stream)) {
+    throw new ERR_INVALID_ARG_VALUE('stream', stream, 'must be writable');
+  }
+
+  const composedStream = staticCompose(this, stream);
+
+  if (options?.signal) {
+    // Not validating as we already validated before
+    addAbortSignalNoValidate(
+      options.signal,
+      composedStream,
+    );
+  }
+
+  return composedStream;
+}
+
+function map(fn, options) {
+  if (typeof fn !== 'function') {
+    throw new ERR_INVALID_ARG_TYPE(
+      'fn', ['Function', 'AsyncFunction'], fn);
+  }
+  if (options != null) {
+    validateObject(options, 'options');
+  }
+  if (options?.signal != null) {
+    validateAbortSignal(options.signal, 'options.signal');
+  }
+
+  let concurrency = 1;
+  if (options?.concurrency != null) {
+    concurrency = MathFloor(options.concurrency);
+  }
+
+  let highWaterMark = concurrency - 1;
+  if (options?.highWaterMark != null) {
+    highWaterMark = MathFloor(options.highWaterMark);
+  }
+
+  validateInteger(concurrency, 'options.concurrency', 1);
+  validateInteger(highWaterMark, 'options.highWaterMark', 0);
+
+  highWaterMark += concurrency;
+
+  return async function* map() {
+    const signal = AbortSignal.any([options?.signal].filter(Boolean));
+    const stream = this;
+    const queue = [];
+    const signalOpt = { signal };
+
+    let next;
+    let resume;
+    let done = false;
+    let cnt = 0;
+
+    function onCatch() {
+      done = true;
+      afterItemProcessed();
+    }
+
+    function afterItemProcessed() {
+      cnt -= 1;
+      maybeResume();
+    }
+
+    function maybeResume() {
+      if (
+        resume &&
+        !done &&
+        cnt < concurrency &&
+        queue.length < highWaterMark
+      ) {
+        resume();
+        resume = null;
+      }
+    }
+
+    async function pump() {
+      try {
+        for await (let val of stream) {
+          if (done) {
+            return;
+          }
+
+          if (signal.aborted) {
+            throw new AbortError();
+          }
+
+          try {
+            val = fn(val, signalOpt);
+
+            if (val === kEmpty) {
+              continue;
+            }
+
+            val = PromiseResolve(val);
+          } catch (err) {
+            val = PromiseReject(err);
+          }
+
+          cnt += 1;
+
+          PromisePrototypeThen(val, afterItemProcessed, onCatch);
+
+          queue.push(val);
+          if (next) {
+            next();
+            next = null;
+          }
+
+          if (!done && (queue.length >= highWaterMark || cnt >= concurrency)) {
+            await new Promise((resolve) => {
+              resume = resolve;
+            });
+          }
+        }
+        queue.push(kEof);
+      } catch (err) {
+        const val = PromiseReject(err);
+        PromisePrototypeThen(val, afterItemProcessed, onCatch);
+        queue.push(val);
+      } finally {
+        done = true;
+        if (next) {
+          next();
+          next = null;
+        }
+      }
+    }
+
+    pump();
+
+    try {
+      while (true) {
+        while (queue.length > 0) {
+          const val = await queue[0];
+
+          if (val === kEof) {
+            return;
+          }
+
+          if (signal.aborted) {
+            throw new AbortError();
+          }
+
+          if (val !== kEmpty) {
+            yield val;
+          }
+
+          queue.shift();
+          maybeResume();
+        }
+
+        await new Promise((resolve) => {
+          next = resolve;
+        });
+      }
+    } finally {
+      done = true;
+      if (resume) {
+        resume();
+        resume = null;
+      }
+    }
+  }.call(this);
+}
+
+async function some(fn, options = undefined) {
+  for await (const unused of filter.call(this, fn, options)) {
+    return true;
+  }
+  return false;
+}
+
+async function every(fn, options = undefined) {
+  if (typeof fn !== 'function') {
+    throw new ERR_INVALID_ARG_TYPE(
+      'fn', ['Function', 'AsyncFunction'], fn);
+  }
+  // https://en.wikipedia.org/wiki/De_Morgan%27s_laws
+  return !(await some.call(this, async (...args) => {
+    return !(await fn(...args));
+  }, options));
+}
+
+async function find(fn, options) {
+  for await (const result of filter.call(this, fn, options)) {
+    return result;
+  }
+  return undefined;
+}
+
+async function forEach(fn, options) {
+  if (typeof fn !== 'function') {
+    throw new ERR_INVALID_ARG_TYPE(
+      'fn', ['Function', 'AsyncFunction'], fn);
+  }
+  async function forEachFn(value, options) {
+    await fn(value, options);
+    return kEmpty;
+  }
+  // eslint-disable-next-line no-unused-vars
+  for await (const unused of map.call(this, forEachFn, options));
+}
+
+function filter(fn, options) {
+  if (typeof fn !== 'function') {
+    throw new ERR_INVALID_ARG_TYPE(
+      'fn', ['Function', 'AsyncFunction'], fn);
+  }
+  async function filterFn(value, options) {
+    if (await fn(value, options)) {
+      return value;
+    }
+    return kEmpty;
+  }
+  return map.call(this, filterFn, options);
+}
+
+// Specific to provide better error to reduce since the argument is only
+// missing if the stream has no items in it - but the code is still appropriate
+class ReduceAwareErrMissingArgs extends ERR_MISSING_ARGS {
+  constructor() {
+    super('reduce');
+    this.message = 'Reduce of an empty stream requires an initial value';
+  }
+}
+
+async function reduce(reducer, initialValue, options) {
+  if (typeof reducer !== 'function') {
+    throw new ERR_INVALID_ARG_TYPE(
+      'reducer', ['Function', 'AsyncFunction'], reducer);
+  }
+  if (options != null) {
+    validateObject(options, 'options');
+  }
+  if (options?.signal != null) {
+    validateAbortSignal(options.signal, 'options.signal');
+  }
+
+  let hasInitialValue = arguments.length > 1;
+  if (options?.signal?.aborted) {
+    const err = new AbortError(undefined, { cause: options.signal.reason });
+    this.once('error', () => {}); // The error is already propagated
+    await finished(this.destroy(err));
+    throw err;
+  }
+  const ac = new AbortController();
+  const signal = ac.signal;
+  if (options?.signal) {
+    const opts = { once: true, [kWeakHandler]: this, [kResistStopPropagation]: true };
+    options.signal.addEventListener('abort', () => ac.abort(), opts);
+  }
+  let gotAnyItemFromStream = false;
+  try {
+    for await (const value of this) {
+      gotAnyItemFromStream = true;
+      if (options?.signal?.aborted) {
+        throw new AbortError();
+      }
+      if (!hasInitialValue) {
+        initialValue = value;
+        hasInitialValue = true;
+      } else {
+        initialValue = await reducer(initialValue, value, { signal });
+      }
+    }
+    if (!gotAnyItemFromStream && !hasInitialValue) {
+      throw new ReduceAwareErrMissingArgs();
+    }
+  } finally {
+    ac.abort();
+  }
+  return initialValue;
+}
+
+async function toArray(options) {
+  if (options != null) {
+    validateObject(options, 'options');
+  }
+  if (options?.signal != null) {
+    validateAbortSignal(options.signal, 'options.signal');
+  }
+
+  const result = [];
+  for await (const val of this) {
+    if (options?.signal?.aborted) {
+      throw new AbortError(undefined, { cause: options.signal.reason });
+    }
+    ArrayPrototypePush(result, val);
+  }
+  return result;
+}
+
+function flatMap(fn, options) {
+  const values = map.call(this, fn, options);
+  return async function* flatMap() {
+    for await (const val of values) {
+      yield* val;
+    }
+  }.call(this);
+}
+
+function toIntegerOrInfinity(number) {
+  // We coerce here to align with the spec
+  // https://github.com/tc39/proposal-iterator-helpers/issues/169
+  number = Number(number);
+  if (NumberIsNaN(number)) {
+    return 0;
+  }
+  if (number < 0) {
+    throw new ERR_OUT_OF_RANGE('number', '>= 0', number);
+  }
+  return number;
+}
+
+function drop(number, options = undefined) {
+  if (options != null) {
+    validateObject(options, 'options');
+  }
+  if (options?.signal != null) {
+    validateAbortSignal(options.signal, 'options.signal');
+  }
+
+  number = toIntegerOrInfinity(number);
+  return async function* drop() {
+    if (options?.signal?.aborted) {
+      throw new AbortError();
+    }
+    for await (const val of this) {
+      if (options?.signal?.aborted) {
+        throw new AbortError();
+      }
+      if (number-- <= 0) {
+        yield val;
+      }
+    }
+  }.call(this);
+}
+
+function take(number, options = undefined) {
+  if (options != null) {
+    validateObject(options, 'options');
+  }
+  if (options?.signal != null) {
+    validateAbortSignal(options.signal, 'options.signal');
+  }
+
+  number = toIntegerOrInfinity(number);
+  return async function* take() {
+    if (options?.signal?.aborted) {
+      throw new AbortError();
+    }
+    for await (const val of this) {
+      if (options?.signal?.aborted) {
+        throw new AbortError();
+      }
+      if (number-- > 0) {
+        yield val;
+      }
+
+      // Don't get another item from iterator in case we reached the end
+      if (number <= 0) {
+        return;
+      }
+    }
+  }.call(this);
+}
+
+module.exports.streamReturningOperators = {
+  drop,
+  filter,
+  flatMap,
+  map,
+  take,
+  compose,
+};
+
+module.exports.promiseReturningOperators = {
+  every,
+  forEach,
+  reduce,
+  toArray,
+  some,
+  find,
+};

lib/internal/streams/passthrough.js

@@ -0,0 +1,47 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+// a passthrough stream.
+// basically just the most minimal sort of Transform stream.
+// Every written chunk gets output as-is.
+
+'use strict';
+
+const {
+  ObjectSetPrototypeOf,
+} = primordials;
+
+module.exports = PassThrough;
+
+const Transform = require('internal/streams/transform');
+ObjectSetPrototypeOf(PassThrough.prototype, Transform.prototype);
+ObjectSetPrototypeOf(PassThrough, Transform);
+
+function PassThrough(options) {
+  if (!(this instanceof PassThrough))
+    return new PassThrough(options);
+
+  Transform.call(this, options);
+}
+
+PassThrough.prototype._transform = function(chunk, encoding, cb) {
+  cb(null, chunk);
+};

lib/internal/streams/pipeline.js

@@ -0,0 +1,472 @@
+// Ported from https://github.com/mafintosh/pump with
+// permission from the author, Mathias Buus (@mafintosh).
+
+'use strict';
+
+const {
+  ArrayIsArray,
+  Promise,
+  SymbolAsyncIterator,
+  SymbolDispose,
+} = 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 {
+  AbortError,
+  aggregateTwoErrors,
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+    ERR_INVALID_RETURN_VALUE,
+    ERR_MISSING_ARGS,
+    ERR_STREAM_DESTROYED,
+    ERR_STREAM_PREMATURE_CLOSE,
+    ERR_STREAM_UNABLE_TO_PIPE,
+  },
+} = require('internal/errors');
+
+const {
+  validateFunction,
+  validateAbortSignal,
+} = require('internal/validators');
+
+const {
+  isIterable,
+  isReadable,
+  isReadableNodeStream,
+  isNodeStream,
+  isTransformStream,
+  isWebStream,
+  isReadableStream,
+  isReadableFinished,
+} = require('internal/streams/utils');
+const { AbortController } = require('internal/abort_controller');
+
+let PassThrough;
+let Readable;
+let addAbortListener;
+
+function destroyer(stream, reading, writing) {
+  let finished = false;
+  stream.on('close', () => {
+    finished = true;
+  });
+
+  const cleanup = eos(stream, { readable: reading, writable: writing }, (err) => {
+    finished = !err;
+  });
+
+  return {
+    destroy: (err) => {
+      if (finished) return;
+      finished = true;
+      destroyImpl.destroyer(stream, err || new ERR_STREAM_DESTROYED('pipe'));
+    },
+    cleanup,
+  };
+}
+
+function popCallback(streams) {
+  // Streams should never be an empty array. It should always contain at least
+  // a single stream. Therefore optimize for the average case instead of
+  // checking for length === 0 as well.
+  validateFunction(streams[streams.length - 1], 'streams[stream.length - 1]');
+  return streams.pop();
+}
+
+function makeAsyncIterable(val) {
+  if (isIterable(val)) {
+    return val;
+  } else if (isReadableNodeStream(val)) {
+    // Legacy streams are not Iterable.
+    return fromReadable(val);
+  }
+  throw new ERR_INVALID_ARG_TYPE(
+    'val', ['Readable', 'Iterable', 'AsyncIterable'], val);
+}
+
+async function* fromReadable(val) {
+  Readable ??= require('internal/streams/readable');
+  yield* Readable.prototype[SymbolAsyncIterator].call(val);
+}
+
+async function pumpToNode(iterable, writable, finish, { end }) {
+  let error;
+  let onresolve = null;
+
+  const resume = (err) => {
+    if (err) {
+      error = err;
+    }
+
+    if (onresolve) {
+      const callback = onresolve;
+      onresolve = null;
+      callback();
+    }
+  };
+
+  const wait = () => new Promise((resolve, reject) => {
+    if (error) {
+      reject(error);
+    } else {
+      onresolve = () => {
+        if (error) {
+          reject(error);
+        } else {
+          resolve();
+        }
+      };
+    }
+  });
+
+  writable.on('drain', resume);
+  const cleanup = eos(writable, { readable: false }, resume);
+
+  try {
+    if (writable.writableNeedDrain) {
+      await wait();
+    }
+
+    for await (const chunk of iterable) {
+      if (!writable.write(chunk)) {
+        await wait();
+      }
+    }
+
+    if (end) {
+      writable.end();
+      await wait();
+    }
+
+    finish();
+  } catch (err) {
+    finish(error !== err ? aggregateTwoErrors(error, err) : err);
+  } finally {
+    cleanup();
+    writable.off('drain', resume);
+  }
+}
+
+async function pumpToWeb(readable, writable, finish, { end }) {
+  if (isTransformStream(writable)) {
+    writable = writable.writable;
+  }
+  // https://streams.spec.whatwg.org/#example-manual-write-with-backpressure
+  const writer = writable.getWriter();
+  try {
+    for await (const chunk of readable) {
+      await writer.ready;
+      writer.write(chunk).catch(() => {});
+    }
+
+    await writer.ready;
+
+    if (end) {
+      await writer.close();
+    }
+
+    finish();
+  } catch (err) {
+    try {
+      await writer.abort(err);
+      finish(err);
+    } catch (err) {
+      finish(err);
+    }
+  }
+}
+
+function pipeline(...streams) {
+  return pipelineImpl(streams, once(popCallback(streams)));
+}
+
+function pipelineImpl(streams, callback, opts) {
+  if (streams.length === 1 && ArrayIsArray(streams[0])) {
+    streams = streams[0];
+  }
+
+  if (streams.length < 2) {
+    throw new ERR_MISSING_ARGS('streams');
+  }
+
+  const ac = new AbortController();
+  const signal = ac.signal;
+  const outerSignal = opts?.signal;
+
+  // Need to cleanup event listeners if last stream is readable
+  // https://github.com/nodejs/node/issues/35452
+  const lastStreamCleanup = [];
+
+  validateAbortSignal(outerSignal, 'options.signal');
+
+  function abort() {
+    finishImpl(new AbortError(undefined, { cause: outerSignal?.reason }));
+  }
+
+  addAbortListener ??= require('internal/events/abort_listener').addAbortListener;
+  let disposable;
+  if (outerSignal) {
+    disposable = addAbortListener(outerSignal, abort);
+  }
+
+  let error;
+  let value;
+  const destroys = [];
+
+  let finishCount = 0;
+
+  function finish(err) {
+    finishImpl(err, --finishCount === 0);
+  }
+
+  function finishOnlyHandleError(err) {
+    finishImpl(err, false);
+  }
+
+  function finishImpl(err, final) {
+    if (err && (!error || error.code === 'ERR_STREAM_PREMATURE_CLOSE')) {
+      error = err;
+    }
+
+    if (!error && !final) {
+      return;
+    }
+
+    while (destroys.length) {
+      destroys.shift()(error);
+    }
+
+    disposable?.[SymbolDispose]();
+    ac.abort();
+
+    if (final) {
+      if (!error) {
+        lastStreamCleanup.forEach((fn) => fn());
+      }
+      process.nextTick(callback, error, value);
+    }
+  }
+
+  let ret;
+  for (let i = 0; i < streams.length; i++) {
+    const stream = streams[i];
+    const reading = i < streams.length - 1;
+    const writing = i > 0;
+    const next = i + 1 < streams.length ? streams[i + 1] : null;
+    const end = reading || opts?.end !== false;
+    const isLastStream = i === streams.length - 1;
+
+    if (isNodeStream(stream)) {
+      if (next !== null && (next?.closed || next?.destroyed)) {
+        throw new ERR_STREAM_UNABLE_TO_PIPE();
+      }
+
+      if (end) {
+        const { destroy, cleanup } = destroyer(stream, reading, writing);
+        destroys.push(destroy);
+
+        if (isReadable(stream) && isLastStream) {
+          lastStreamCleanup.push(cleanup);
+        }
+      }
+
+      // Catch stream errors that occur after pipe/pump has completed.
+      function onError(err) {
+        if (
+          err &&
+          err.name !== 'AbortError' &&
+          err.code !== 'ERR_STREAM_PREMATURE_CLOSE'
+        ) {
+          finishOnlyHandleError(err);
+        }
+      }
+      stream.on('error', onError);
+      if (isReadable(stream) && isLastStream) {
+        lastStreamCleanup.push(() => {
+          stream.removeListener('error', onError);
+        });
+      }
+    }
+
+    if (i === 0) {
+      if (typeof stream === 'function') {
+        ret = stream({ signal });
+        if (!isIterable(ret)) {
+          throw new ERR_INVALID_RETURN_VALUE(
+            'Iterable, AsyncIterable or Stream', 'source', ret);
+        }
+      } else if (isIterable(stream) || isReadableNodeStream(stream) || isTransformStream(stream)) {
+        ret = stream;
+      } else {
+        ret = Duplex.from(stream);
+      }
+    } else if (typeof stream === 'function') {
+      if (isTransformStream(ret)) {
+        ret = makeAsyncIterable(ret?.readable);
+      } else {
+        ret = makeAsyncIterable(ret);
+      }
+      ret = stream(ret, { signal });
+
+      if (reading) {
+        if (!isIterable(ret, true)) {
+          throw new ERR_INVALID_RETURN_VALUE(
+            'AsyncIterable', `transform[${i - 1}]`, ret);
+        }
+      } else {
+        PassThrough ??= require('internal/streams/passthrough');
+
+        // If the last argument to pipeline is not a stream
+        // we must create a proxy stream so that pipeline(...)
+        // always returns a stream which can be further
+        // composed through `.pipe(stream)`.
+
+        const pt = new PassThrough({
+          objectMode: true,
+        });
+
+        // Handle Promises/A+ spec, `then` could be a getter that throws on
+        // second use.
+        const then = ret?.then;
+        if (typeof then === 'function') {
+          finishCount++;
+          then.call(ret,
+                    (val) => {
+                      value = val;
+                      if (val != null) {
+                        pt.write(val);
+                      }
+                      if (end) {
+                        pt.end();
+                      }
+                      process.nextTick(finish);
+                    }, (err) => {
+                      pt.destroy(err);
+                      process.nextTick(finish, err);
+                    },
+          );
+        } else if (isIterable(ret, true)) {
+          finishCount++;
+          pumpToNode(ret, pt, finish, { end });
+        } else if (isReadableStream(ret) || isTransformStream(ret)) {
+          const toRead = ret.readable || ret;
+          finishCount++;
+          pumpToNode(toRead, pt, finish, { end });
+        } else {
+          throw new ERR_INVALID_RETURN_VALUE(
+            'AsyncIterable or Promise', 'destination', ret);
+        }
+
+        ret = pt;
+
+        const { destroy, cleanup } = destroyer(ret, false, true);
+        destroys.push(destroy);
+        if (isLastStream) {
+          lastStreamCleanup.push(cleanup);
+        }
+      }
+    } else if (isNodeStream(stream)) {
+      if (isReadableNodeStream(ret)) {
+        finishCount += 2;
+        const cleanup = pipe(ret, stream, finish, finishOnlyHandleError, { end });
+        if (isReadable(stream) && isLastStream) {
+          lastStreamCleanup.push(cleanup);
+        }
+      } else if (isTransformStream(ret) || isReadableStream(ret)) {
+        const toRead = ret.readable || ret;
+        finishCount++;
+        pumpToNode(toRead, stream, finish, { end });
+      } else if (isIterable(ret)) {
+        finishCount++;
+        pumpToNode(ret, stream, finish, { end });
+      } else {
+        throw new ERR_INVALID_ARG_TYPE(
+          'val', ['Readable', 'Iterable', 'AsyncIterable', 'ReadableStream', 'TransformStream'], ret);
+      }
+      ret = stream;
+    } else if (isWebStream(stream)) {
+      if (isReadableNodeStream(ret)) {
+        finishCount++;
+        pumpToWeb(makeAsyncIterable(ret), stream, finish, { end });
+      } else if (isReadableStream(ret) || isIterable(ret)) {
+        finishCount++;
+        pumpToWeb(ret, stream, finish, { end });
+      } else if (isTransformStream(ret)) {
+        finishCount++;
+        pumpToWeb(ret.readable, stream, finish, { end });
+      } else {
+        throw new ERR_INVALID_ARG_TYPE(
+          'val', ['Readable', 'Iterable', 'AsyncIterable', 'ReadableStream', 'TransformStream'], ret);
+      }
+      ret = stream;
+    } else {
+      ret = Duplex.from(stream);
+    }
+  }
+
+  if (signal?.aborted || outerSignal?.aborted) {
+    process.nextTick(abort);
+  }
+
+  return ret;
+}
+
+function pipe(src, dst, finish, finishOnlyHandleError, { end }) {
+  let ended = false;
+  dst.on('close', () => {
+    if (!ended) {
+      // Finish if the destination closes before the source has completed.
+      finishOnlyHandleError(new ERR_STREAM_PREMATURE_CLOSE());
+    }
+  });
+
+  src.pipe(dst, { end: false }); // If end is true we already will have a listener to end dst.
+
+  if (end) {
+    // Compat. Before node v10.12.0 stdio used to throw an error so
+    // pipe() did/does not end() stdio destinations.
+    // Now they allow it but "secretly" don't close the underlying fd.
+
+    function endFn() {
+      ended = true;
+      dst.end();
+    }
+
+    if (isReadableFinished(src)) { // End the destination if the source has already ended.
+      process.nextTick(endFn);
+    } else {
+      src.once('end', endFn);
+    }
+  } else {
+    finish();
+  }
+
+  eos(src, { readable: true, writable: false }, (err) => {
+    const rState = src._readableState;
+    if (
+      err &&
+      err.code === 'ERR_STREAM_PREMATURE_CLOSE' &&
+      (rState?.ended && !rState.errored && !rState.errorEmitted)
+    ) {
+      // Some readable streams will emit 'close' before 'end'. However, since
+      // this is on the readable side 'end' should still be emitted if the
+      // stream has been ended and no error emitted. This should be allowed in
+      // favor of backwards compatibility. Since the stream is piped to a
+      // destination this should not result in any observable difference.
+      // We don't need to check if this is a writable premature close since
+      // eos will only fail with premature close on the reading side for
+      // duplex streams.
+      src
+        .once('end', finish)
+        .once('error', finish);
+    } else {
+      finish(err);
+    }
+  });
+  return eos(dst, { readable: false, writable: true }, finish);
+}
+
+module.exports = { pipelineImpl, pipeline };

lib/internal/streams/state.js

@@ -0,0 +1,51 @@
+'use strict';
+
+const {
+  MathFloor,
+  NumberIsInteger,
+} = primordials;
+const { validateInteger } = require('internal/validators');
+
+const { ERR_INVALID_ARG_VALUE } = require('internal/errors').codes;
+
+// TODO (fix): For some reason Windows CI fails with bigger hwm.
+let defaultHighWaterMarkBytes = process.platform === 'win32' ? 16 * 1024 : 64 * 1024;
+let defaultHighWaterMarkObjectMode = 16;
+
+function highWaterMarkFrom(options, isDuplex, duplexKey) {
+  return options.highWaterMark != null ? options.highWaterMark :
+    isDuplex ? options[duplexKey] : null;
+}
+
+function getDefaultHighWaterMark(objectMode) {
+  return objectMode ? defaultHighWaterMarkObjectMode : defaultHighWaterMarkBytes;
+}
+
+function setDefaultHighWaterMark(objectMode, value) {
+  validateInteger(value, 'value', 0);
+  if (objectMode) {
+    defaultHighWaterMarkObjectMode = value;
+  } else {
+    defaultHighWaterMarkBytes = value;
+  }
+}
+
+function getHighWaterMark(state, options, duplexKey, isDuplex) {
+  const hwm = highWaterMarkFrom(options, isDuplex, duplexKey);
+  if (hwm != null) {
+    if (!NumberIsInteger(hwm) || hwm < 0) {
+      const name = isDuplex ? `options.${duplexKey}` : 'options.highWaterMark';
+      throw new ERR_INVALID_ARG_VALUE(name, hwm);
+    }
+    return MathFloor(hwm);
+  }
+
+  // Default value
+  return getDefaultHighWaterMark(state.objectMode);
+}
+
+module.exports = {
+  getHighWaterMark,
+  getDefaultHighWaterMark,
+  setDefaultHighWaterMark,
+};

lib/internal/streams/transform.js

@@ -0,0 +1,204 @@
+// Copyright Joyent, Inc. and other Node contributors.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a
+// copy of this software and associated documentation files (the
+// "Software"), to deal in the Software without restriction, including
+// without limitation the rights to use, copy, modify, merge, publish,
+// distribute, sublicense, and/or sell copies of the Software, and to permit
+// persons to whom the Software is furnished to do so, subject to the
+// following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+// NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+// OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+// USE OR OTHER DEALINGS IN THE SOFTWARE.
+
+// a transform stream is a readable/writable stream where you do
+// something with the data.  Sometimes it's called a "filter",
+// but that's not a great name for it, since that implies a thing where
+// some bits pass through, and others are simply ignored.  (That would
+// be a valid example of a transform, of course.)
+//
+// While the output is causally related to the input, it's not a
+// necessarily symmetric or synchronous transformation.  For example,
+// a zlib stream might take multiple plain-text writes(), and then
+// emit a single compressed chunk some time in the future.
+//
+// Here's how this works:
+//
+// The Transform stream has all the aspects of the readable and writable
+// stream classes.  When you write(chunk), that calls _write(chunk,cb)
+// internally, and returns false if there's a lot of pending writes
+// buffered up.  When you call read(), that calls _read(n) until
+// there's enough pending readable data buffered up.
+//
+// In a transform stream, the written data is placed in a buffer.  When
+// _read(n) is called, it transforms the queued up data, calling the
+// buffered _write cb's as it consumes chunks.  If consuming a single
+// written chunk would result in multiple output chunks, then the first
+// outputted bit calls the readcb, and subsequent chunks just go into
+// the read buffer, and will cause it to emit 'readable' if necessary.
+//
+// This way, back-pressure is actually determined by the reading side,
+// since _read has to be called to start processing a new chunk.  However,
+// a pathological inflate type of transform can cause excessive buffering
+// here.  For example, imagine a stream where every byte of input is
+// interpreted as an integer from 0-255, and then results in that many
+// bytes of output.  Writing the 4 bytes {ff,ff,ff,ff} would result in
+// 1kb of data being output.  In this case, you could write a very small
+// amount of input, and end up with a very large amount of output.  In
+// such a pathological inflating mechanism, there'd be no way to tell
+// the system to stop doing the transform.  A single 4MB write could
+// cause the system to run out of memory.
+//
+// However, even in such a pathological case, only a single written chunk
+// would be consumed, and then the rest would wait (un-transformed) until
+// the results of the previous transformed chunk were consumed.
+
+'use strict';
+
+const {
+  ObjectSetPrototypeOf,
+  Symbol,
+} = primordials;
+
+module.exports = Transform;
+const {
+  ERR_METHOD_NOT_IMPLEMENTED,
+} = require('internal/errors').codes;
+const Duplex = require('internal/streams/duplex');
+const { getHighWaterMark } = require('internal/streams/state');
+ObjectSetPrototypeOf(Transform.prototype, Duplex.prototype);
+ObjectSetPrototypeOf(Transform, Duplex);
+
+const kCallback = Symbol('kCallback');
+
+function Transform(options) {
+  if (!(this instanceof Transform))
+    return new Transform(options);
+
+  // TODO (ronag): This should preferably always be
+  // applied but would be semver-major. Or even better;
+  // make Transform a Readable with the Writable interface.
+  const readableHighWaterMark = options ? getHighWaterMark(this, options, 'readableHighWaterMark', true) : null;
+  if (readableHighWaterMark === 0) {
+    // A Duplex will buffer both on the writable and readable side while
+    // a Transform just wants to buffer hwm number of elements. To avoid
+    // buffering twice we disable buffering on the writable side.
+    options = {
+      ...options,
+      highWaterMark: null,
+      readableHighWaterMark,
+      writableHighWaterMark: options.writableHighWaterMark || 0,
+    };
+  }
+
+  Duplex.call(this, options);
+
+  // We have implemented the _read method, and done the other things
+  // that Readable wants before the first _read call, so unset the
+  // sync guard flag.
+  this._readableState.sync = false;
+
+  this[kCallback] = null;
+
+  if (options) {
+    if (typeof options.transform === 'function')
+      this._transform = options.transform;
+
+    if (typeof options.flush === 'function')
+      this._flush = options.flush;
+  }
+
+  // When the writable side finishes, then flush out anything remaining.
+  // Backwards compat. Some Transform streams incorrectly implement _final
+  // instead of or in addition to _flush. By using 'prefinish' instead of
+  // implementing _final we continue supporting this unfortunate use case.
+  this.on('prefinish', prefinish);
+}
+
+function final(cb) {
+  if (typeof this._flush === 'function' && !this.destroyed) {
+    this._flush((er, data) => {
+      if (er) {
+        if (cb) {
+          cb(er);
+        } else {
+          this.destroy(er);
+        }
+        return;
+      }
+
+      if (data != null) {
+        this.push(data);
+      }
+      this.push(null);
+      if (cb) {
+        cb();
+      }
+    });
+  } else {
+    this.push(null);
+    if (cb) {
+      cb();
+    }
+  }
+}
+
+function prefinish() {
+  if (this._final !== final) {
+    final.call(this);
+  }
+}
+
+Transform.prototype._final = final;
+
+Transform.prototype._transform = function(chunk, encoding, callback) {
+  throw new ERR_METHOD_NOT_IMPLEMENTED('_transform()');
+};
+
+Transform.prototype._write = function(chunk, encoding, callback) {
+  const rState = this._readableState;
+  const wState = this._writableState;
+  const length = rState.length;
+
+  this._transform(chunk, encoding, (err, val) => {
+    if (err) {
+      callback(err);
+      return;
+    }
+
+    if (val != null) {
+      this.push(val);
+    }
+
+    if (rState.ended) {
+      // If user has called this.push(null) we have to
+      // delay the callback to properly propagate the new
+      // state.
+      process.nextTick(callback);
+    } else if (
+      wState.ended || // Backwards compat.
+      length === rState.length || // Backwards compat.
+      rState.length < rState.highWaterMark
+    ) {
+      callback();
+    } else {
+      this[kCallback] = callback;
+    }
+  });
+};
+
+Transform.prototype._read = function() {
+  if (this[kCallback]) {
+    const callback = this[kCallback];
+    this[kCallback] = null;
+    callback();
+  }
+};

lib/internal/streams/utils.js

@@ -0,0 +1,363 @@
+'use strict';
+
+const {
+  Symbol,
+  SymbolAsyncIterator,
+  SymbolFor,
+  SymbolIterator,
+} = primordials;
+
+// We need to use SymbolFor to make these globally available
+// for interoperability with readable-stream, i.e. readable-stream
+// and node core needs to be able to read/write private state
+// from each other for proper interoperability.
+const kIsDestroyed = SymbolFor('nodejs.stream.destroyed');
+const kIsErrored = SymbolFor('nodejs.stream.errored');
+const kIsReadable = SymbolFor('nodejs.stream.readable');
+const kIsWritable = SymbolFor('nodejs.stream.writable');
+const kIsDisturbed = SymbolFor('nodejs.stream.disturbed');
+
+const kOnConstructed = Symbol('kOnConstructed');
+
+const kIsClosedPromise = SymbolFor('nodejs.webstream.isClosedPromise');
+const kControllerErrorFunction = SymbolFor('nodejs.webstream.controllerErrorFunction');
+
+const kState = Symbol('kState');
+const kObjectMode = 1 << 0;
+const kErrorEmitted = 1 << 1;
+const kAutoDestroy = 1 << 2;
+const kEmitClose = 1 << 3;
+const kDestroyed = 1 << 4;
+const kClosed = 1 << 5;
+const kCloseEmitted = 1 << 6;
+const kErrored = 1 << 7;
+const kConstructed = 1 << 8;
+
+function isReadableNodeStream(obj, strict = false) {
+  return !!(
+    obj &&
+    typeof obj.pipe === 'function' &&
+    typeof obj.on === 'function' &&
+    (
+      !strict ||
+      (typeof obj.pause === 'function' && typeof obj.resume === '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 isDuplexNodeStream(obj) {
+  return !!(
+    obj &&
+    (typeof obj.pipe === 'function' && obj._readableState) &&
+    typeof obj.on === 'function' &&
+    typeof obj.write === 'function'
+  );
+}
+
+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 isReadableStream(obj) {
+  return !!(
+    obj &&
+    !isNodeStream(obj) &&
+    typeof obj.pipeThrough === 'function' &&
+    typeof obj.getReader === 'function' &&
+    typeof obj.cancel === 'function'
+  );
+}
+
+function isWritableStream(obj) {
+  return !!(
+    obj &&
+    !isNodeStream(obj) &&
+    typeof obj.getWriter === 'function' &&
+    typeof obj.abort === 'function'
+  );
+}
+
+function isTransformStream(obj) {
+  return !!(
+    obj &&
+    !isNodeStream(obj) &&
+    typeof obj.readable === 'object' &&
+    typeof obj.writable === 'object'
+  );
+}
+
+function isWebStream(obj) {
+  return isReadableStream(obj) || isWritableStream(obj) || isTransformStream(obj);
+}
+
+function isIterable(obj, isAsync) {
+  if (obj == null) return false;
+  if (isAsync === true) return typeof obj[SymbolAsyncIterator] === 'function';
+  if (isAsync === false) return typeof obj[SymbolIterator] === 'function';
+  return typeof obj[SymbolAsyncIterator] === 'function' ||
+    typeof obj[SymbolIterator] === 'function';
+}
+
+function isDestroyed(stream) {
+  if (!isNodeStream(stream)) return null;
+  const wState = stream._writableState;
+  const rState = stream._readableState;
+  const state = wState || rState;
+  return !!(stream.destroyed || stream[kIsDestroyed] || state?.destroyed);
+}
+
+// Have been end():d.
+function isWritableEnded(stream) {
+  if (!isWritableNodeStream(stream)) return null;
+  if (stream.writableEnded === true) return true;
+  const wState = stream._writableState;
+  if (wState?.errored) return false;
+  if (typeof wState?.ended !== 'boolean') return null;
+  return wState.ended;
+}
+
+// Have emitted 'finish'.
+function isWritableFinished(stream, strict) {
+  if (!isWritableNodeStream(stream)) return null;
+  if (stream.writableFinished === true) return true;
+  const wState = stream._writableState;
+  if (wState?.errored) return false;
+  if (typeof wState?.finished !== 'boolean') return null;
+  return !!(
+    wState.finished ||
+    (strict === false && wState.ended === true && wState.length === 0)
+  );
+}
+
+// Have been push(null):d.
+function isReadableEnded(stream) {
+  if (!isReadableNodeStream(stream)) return null;
+  if (stream.readableEnded === true) return true;
+  const rState = stream._readableState;
+  if (!rState || rState.errored) return false;
+  if (typeof rState?.ended !== 'boolean') return null;
+  return rState.ended;
+}
+
+// Have emitted 'end'.
+function isReadableFinished(stream, strict) {
+  if (!isReadableNodeStream(stream)) return null;
+  const rState = stream._readableState;
+  if (rState?.errored) return false;
+  if (typeof rState?.endEmitted !== 'boolean') return null;
+  return !!(
+    rState.endEmitted ||
+    (strict === false && rState.ended === true && rState.length === 0)
+  );
+}
+
+function isReadable(stream) {
+  if (stream && stream[kIsReadable] != null) return stream[kIsReadable];
+  if (typeof stream?.readable !== 'boolean') return null;
+  if (isDestroyed(stream)) return false;
+  return isReadableNodeStream(stream) &&
+    stream.readable &&
+    !isReadableFinished(stream);
+}
+
+function isWritable(stream) {
+  if (stream && stream[kIsWritable] != null) return stream[kIsWritable];
+  if (typeof stream?.writable !== 'boolean') return null;
+  if (isDestroyed(stream)) return false;
+  return isWritableNodeStream(stream) &&
+    stream.writable &&
+    !isWritableEnded(stream);
+}
+
+function isFinished(stream, opts) {
+  if (!isNodeStream(stream)) {
+    return null;
+  }
+
+  if (isDestroyed(stream)) {
+    return true;
+  }
+
+  if (opts?.readable !== false && isReadable(stream)) {
+    return false;
+  }
+
+  if (opts?.writable !== false && isWritable(stream)) {
+    return false;
+  }
+
+  return true;
+}
+
+function isWritableErrored(stream) {
+  if (!isNodeStream(stream)) {
+    return null;
+  }
+
+  if (stream.writableErrored) {
+    return stream.writableErrored;
+  }
+
+  return stream._writableState?.errored ?? null;
+}
+
+function isReadableErrored(stream) {
+  if (!isNodeStream(stream)) {
+    return null;
+  }
+
+  if (stream.readableErrored) {
+    return stream.readableErrored;
+  }
+
+  return stream._readableState?.errored ?? null;
+}
+
+function isClosed(stream) {
+  if (!isNodeStream(stream)) {
+    return null;
+  }
+
+  if (typeof stream.closed === 'boolean') {
+    return stream.closed;
+  }
+
+  const wState = stream._writableState;
+  const rState = stream._readableState;
+
+  if (
+    typeof wState?.closed === 'boolean' ||
+    typeof rState?.closed === 'boolean'
+  ) {
+    return wState?.closed || rState?.closed;
+  }
+
+  if (typeof stream._closed === 'boolean' && isOutgoingMessage(stream)) {
+    return stream._closed;
+  }
+
+  return null;
+}
+
+function isOutgoingMessage(stream) {
+  return (
+    typeof stream._closed === 'boolean' &&
+    typeof stream._defaultKeepAlive === 'boolean' &&
+    typeof stream._removedConnection === 'boolean' &&
+    typeof stream._removedContLen === 'boolean'
+  );
+}
+
+function isServerResponse(stream) {
+  return (
+    typeof stream._sent100 === 'boolean' &&
+    isOutgoingMessage(stream)
+  );
+}
+
+function isServerRequest(stream) {
+  return (
+    typeof stream._consuming === 'boolean' &&
+    typeof stream._dumped === 'boolean' &&
+    stream.req?.upgradeOrConnect === undefined
+  );
+}
+
+function willEmitClose(stream) {
+  if (!isNodeStream(stream)) return null;
+
+  const wState = stream._writableState;
+  const rState = stream._readableState;
+  const state = wState || rState;
+
+  return (!state && isServerResponse(stream)) || !!(
+    state?.autoDestroy &&
+    state.emitClose &&
+    state.closed === false
+  );
+}
+
+function isDisturbed(stream) {
+  return !!(stream && (
+    stream[kIsDisturbed] ??
+    (stream.readableDidRead || stream.readableAborted)
+  ));
+}
+
+function isErrored(stream) {
+  return !!(stream && (
+    stream[kIsErrored] ??
+    stream.readableErrored ??
+    stream.writableErrored ??
+    stream._readableState?.errorEmitted ??
+    stream._writableState?.errorEmitted ??
+    stream._readableState?.errored ??
+    stream._writableState?.errored
+  ));
+}
+
+module.exports = {
+  kOnConstructed,
+  isDestroyed,
+  kIsDestroyed,
+  isDisturbed,
+  kIsDisturbed,
+  isErrored,
+  kIsErrored,
+  isReadable,
+  kIsReadable,
+  kIsClosedPromise,
+  kControllerErrorFunction,
+  kIsWritable,
+  isClosed,
+  isDuplexNodeStream,
+  isFinished,
+  isIterable,
+  isReadableNodeStream,
+  isReadableStream,
+  isReadableEnded,
+  isReadableFinished,
+  isReadableErrored,
+  isNodeStream,
+  isWebStream,
+  isWritable,
+  isWritableNodeStream,
+  isWritableStream,
+  isWritableEnded,
+  isWritableFinished,
+  isWritableErrored,
+  isServerRequest,
+  isServerResponse,
+  willEmitClose,
+  isTransformStream,
+  kState,
+  // bitfields
+  kObjectMode,
+  kErrorEmitted,
+  kAutoDestroy,
+  kEmitClose,
+  kDestroyed,
+  kClosed,
+  kCloseEmitted,
+  kErrored,
+  kConstructed,
+};