stream: implement WHATWG streams

Experimental implementation of the WHATWG streams standard.

Signed-off-by: James M Snell <jasnell@gmail.com>

PR-URL: https://github.com/nodejs/node/pull/39062
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Michaël Zasso <targos@protonmail.com>

doc/api/errors.md

@@ -1428,6 +1428,11 @@ is set for the `Http2Stream`.
 `http2.connect()` was passed a URL that uses any protocol other than `http:` or
 `https:`.
 
+<a id="ERR_ILLEGAL_CONSTRUCTOR"></a>
+### `ERR_ILLEGAL_CONSTRUCTOR`
+
+An attempt was made to construct an object using a non-public constructor.
+
 <a id="ERR_INCOMPATIBLE_OPTION_PAIR"></a>
 ### `ERR_INCOMPATIBLE_OPTION_PAIR`
 

doc/api/index.md

@@ -64,6 +64,7 @@
 * [VM](vm.md)
 * [WASI](wasi.md)
 * [Web Crypto API](webcrypto.md)
+* [Web Streams API](webstreams.md)
 * [Worker threads](worker_threads.md)
 * [Zlib](zlib.md)
 

lib/internal/abort_controller.js

@@ -143,6 +143,7 @@ ObjectDefineProperty(AbortController.prototype, SymbolToStringTag, {
 });
 
 module.exports = {
+  kAborted,
   AbortController,
   AbortSignal,
 };

lib/internal/errors.js

@@ -1033,6 +1033,7 @@ E('ERR_HTTP_SOCKET_ENCODING',
   'Changing the socket encoding is not allowed per RFC7230 Section 3.', Error);
 E('ERR_HTTP_TRAILER_INVALID',
   'Trailers are invalid with this transfer encoding', Error);
+E('ERR_ILLEGAL_CONSTRUCTOR', 'Illegal constructor', TypeError);
 E('ERR_INCOMPATIBLE_OPTION_PAIR',
   'Option "%s" cannot be used in combination with option "%s"', TypeError);
 E('ERR_INPUT_TYPE_NOT_ALLOWED', '--input-type can only be used with string ' +
@@ -1256,8 +1257,8 @@ E('ERR_INVALID_RETURN_VALUE', (input, name, value) => {
   }
   return `Expected ${input} to be returned from the "${name}"` +
          ` function but got ${type}.`;
-}, TypeError);
-E('ERR_INVALID_STATE', 'Invalid state: %s', Error);
+}, TypeError, RangeError);
+E('ERR_INVALID_STATE', 'Invalid state: %s', Error, TypeError, RangeError);
 E('ERR_INVALID_SYNC_FORK_INPUT',
   'Asynchronous forks do not support ' +
     'Buffer, TypedArray, DataView or string input: %s',
@@ -1361,7 +1362,7 @@ E('ERR_NO_CRYPTO',
   'Node.js is not compiled with OpenSSL crypto support', Error);
 E('ERR_NO_ICU',
   '%s is not supported on Node.js compiled without ICU', TypeError);
-E('ERR_OPERATION_FAILED', 'Operation failed: %s', Error);
+E('ERR_OPERATION_FAILED', 'Operation failed: %s', Error, TypeError);
 E('ERR_OUT_OF_RANGE',
   (str, range, input, replaceDefaultBoolean = false) => {
     assert(range, 'Missing "range" argument');

lib/internal/per_context/primordials.js

@@ -415,5 +415,10 @@ primordials.SafePromisePrototypeFinally = (thisPromise, onFinally) =>
       .then(a, b)
   );
 
+primordials.AsyncIteratorPrototype =
+  primordials.ReflectGetPrototypeOf(
+    primordials.ReflectGetPrototypeOf(
+      async function* () {}).prototype);
+
 ObjectSetPrototypeOf(primordials, null);
 ObjectFreeze(primordials);

lib/internal/webstreams/queuingstrategies.js

@@ -0,0 +1,168 @@
+'use strict';
+
+const {
+  ObjectDefineProperties,
+  SymbolToStringTag,
+} = primordials;
+
+const {
+  codes: {
+    ERR_INVALID_THIS,
+    ERR_MISSING_OPTION,
+  },
+} = require('internal/errors');
+
+const {
+  customInspectSymbol: kInspect,
+} = require('internal/util');
+
+const {
+  customInspect,
+  isBrandCheck,
+  kType,
+  kState,
+} = require('internal/webstreams/util');
+
+const {
+  validateObject,
+} = require('internal/validators');
+
+const isByteLengthQueuingStrategy =
+  isBrandCheck('ByteLengthQueuingStrategy');
+
+const isCountQueuingStrategy =
+  isBrandCheck('CountQueuingStrategy');
+
+/**
+ * @callback QueuingStrategySize
+ * @param {any} chunk
+ * @returns {number}
+ *
+ * @typedef {{
+ *   highWaterMark : number,
+ *   size? : QueuingStrategySize,
+ * }} QueuingStrategy
+ */
+
+// eslint-disable-next-line func-name-matching,func-style
+const byteSizeFunction = function size(chunk) { return chunk.byteLength; };
+
+// eslint-disable-next-line func-name-matching,func-style
+const countSizeFunction = function size() { return 1; };
+
+/**
+ * @type {QueuingStrategy}
+ */
+class ByteLengthQueuingStrategy {
+  [kType] = 'ByteLengthQueuingStrategy';
+
+  get [SymbolToStringTag]() { return this[kType]; }
+
+  /**
+   * @param {{
+   *   highWaterMark : number
+   * }} init
+   */
+  constructor(init) {
+    validateObject(init, 'init');
+    if (init.highWaterMark === undefined)
+      throw new ERR_MISSING_OPTION('options.highWaterMark');
+
+    // The highWaterMark value is not checked until the strategy
+    // is actually used, per the spec.
+    this[kState] = {
+      highWaterMark: +init.highWaterMark,
+    };
+  }
+
+  /**
+   * @readonly
+   * @type {number}
+   */
+  get highWaterMark() {
+    if (!isByteLengthQueuingStrategy(this))
+      throw new ERR_INVALID_THIS('ByteLengthQueuingStrategy');
+    return this[kState].highWaterMark;
+  }
+
+  /**
+   * @type {QueuingStrategySize}
+   */
+  get size() {
+    if (!isByteLengthQueuingStrategy(this))
+      throw new ERR_INVALID_THIS('ByteLengthQueuingStrategy');
+    return byteSizeFunction;
+  }
+
+  [kInspect](depth, options) {
+    return customInspect(depth, options, this[kType], {
+      highWaterMark: this.highWaterMark,
+    });
+  }
+}
+
+ObjectDefineProperties(ByteLengthQueuingStrategy.prototype, {
+  highWaterMark: { enumerable: true },
+  size: { enumerable: true },
+});
+
+/**
+ * @type {QueuingStrategy}
+ */
+class CountQueuingStrategy {
+  [kType] = 'CountQueuingStrategy';
+
+  get [SymbolToStringTag]() { return this[kType]; }
+
+  /**
+   * @param {{
+   *   highWaterMark : number
+   * }} init
+   */
+  constructor(init) {
+    validateObject(init, 'init');
+    if (init.highWaterMark === undefined)
+      throw new ERR_MISSING_OPTION('options.highWaterMark');
+
+    // The highWaterMark value is not checked until the strategy
+    // is actually used, per the spec.
+    this[kState] = {
+      highWaterMark: +init.highWaterMark,
+    };
+  }
+
+  /**
+   * @readonly
+   * @type {number}
+   */
+  get highWaterMark() {
+    if (!isCountQueuingStrategy(this))
+      throw new ERR_INVALID_THIS('CountQueuingStrategy');
+    return this[kState].highWaterMark;
+  }
+
+  /**
+   * @type {QueuingStrategySize}
+   */
+  get size() {
+    if (!isCountQueuingStrategy(this))
+      throw new ERR_INVALID_THIS('CountQueuingStrategy');
+    return countSizeFunction;
+  }
+
+  [kInspect](depth, options) {
+    return customInspect(depth, options, this[kType], {
+      highWaterMark: this.highWaterMark,
+    });
+  }
+}
+
+ObjectDefineProperties(CountQueuingStrategy.prototype, {
+  highWaterMark: { enumerable: true },
+  size: { enumerable: true },
+});
+
+module.exports = {
+  ByteLengthQueuingStrategy,
+  CountQueuingStrategy,
+};

lib/internal/webstreams/transfer.js

@@ -0,0 +1,299 @@
+'use strict';
+
+const {
+  ObjectDefineProperties,
+  PromiseResolve,
+  ReflectConstruct,
+} = primordials;
+
+const {
+  kState,
+  setPromiseHandled,
+} = require('internal/webstreams/util');
+
+const {
+  DOMException,
+} = internalBinding('messaging');
+
+const {
+  ReadableStream,
+  readableStreamDefaultControllerEnqueue,
+  readableStreamDefaultControllerClose,
+  readableStreamDefaultControllerError,
+  readableStreamPipeTo,
+} = require('internal/webstreams/readablestream');
+
+const {
+  WritableStream,
+  writableStreamDefaultControllerErrorIfNeeded,
+} = require('internal/webstreams/writablestream');
+
+const {
+  createDeferredPromise,
+} = require('internal/util');
+
+const assert = require('internal/assert');
+
+const {
+  makeTransferable,
+  kClone,
+  kDeserialize,
+} = require('internal/worker/js_transferable');
+
+// This class is a bit of a hack. The Node.js implementation of
+// DOMException is not transferable/cloneable. This provides us
+// with a variant that is. Unfortunately, it means playing around
+// a bit with the message, name, and code properties and the
+// prototype. We can revisit this if DOMException is ever made
+// properly cloneable.
+class CloneableDOMException extends DOMException {
+  constructor(message, name) {
+    super(message, name);
+    this[kDeserialize]({
+      message: this.message,
+      name: this.name,
+      code: this.code,
+    });
+    // eslint-disable-next-line no-constructor-return
+    return makeTransferable(this);
+  }
+
+  [kClone]() {
+    return {
+      data: {
+        message: this.message,
+        name: this.name,
+        code: this.code,
+      },
+      deserializeInfo:
+        'internal/webstreams/transfer:InternalCloneableDOMException'
+    };
+  }
+
+  [kDeserialize]({ message, name, code }) {
+    ObjectDefineProperties(this, {
+      message: {
+        configurable: true,
+        enumerable: true,
+        get() { return message; },
+      },
+      name: {
+        configurable: true,
+        enumerable: true,
+        get() { return name; },
+      },
+      code: {
+        configurable: true,
+        enumerable: true,
+        get() { return code; },
+      },
+    });
+  }
+}
+
+function InternalCloneableDOMException() {
+  return makeTransferable(
+    ReflectConstruct(
+      CloneableDOMException,
+      [],
+      DOMException));
+}
+InternalCloneableDOMException[kDeserialize] = () => {};
+
+class CrossRealmTransformReadableSource {
+  constructor(port) {
+    this[kState] = {
+      port,
+      controller: undefined,
+    };
+
+    port.onmessage = ({ data }) => {
+      const {
+        controller,
+      } = this[kState];
+      const {
+        type,
+        value,
+      } = data;
+      switch (type) {
+        case 'chunk':
+          readableStreamDefaultControllerEnqueue(
+            controller,
+            value);
+          break;
+        case 'close':
+          readableStreamDefaultControllerClose(controller);
+          port.close();
+          break;
+        case 'error':
+          readableStreamDefaultControllerError(controller, value);
+          port.close();
+          break;
+      }
+    };
+
+    port.onmessageerror = () => {
+      const error = new CloneableDOMException(
+        'Internal transfered ReadableStream error',
+        'DataCloneError');
+      port.postMessage({ type: 'error', value: error });
+      readableStreamDefaultControllerError(
+        this[kState].controller,
+        error);
+      port.close();
+    };
+  }
+
+  start(controller) {
+    this[kState].controller = controller;
+  }
+
+  async pull() {
+    this[kState].port.postMessage({ type: 'pull' });
+  }
+
+  async cancel(reason) {
+    try {
+      this[kState].port.postMessage({ type: 'error', value: reason });
+    } catch (error) {
+      if (error instanceof DOMException) {
+        // eslint-disable-next-line no-ex-assign
+        error = new CloneableDOMException(error.message, error.name);
+      }
+      this[kState].port.postMessage({ type: 'error', value: error });
+      throw error;
+    } finally {
+      this[kState].port.close();
+    }
+  }
+}
+
+class CrossRealmTransformWritableSink {
+  constructor(port) {
+    this[kState] = {
+      port,
+      controller: undefined,
+      backpressurePromise: createDeferredPromise(),
+    };
+
+    port.onmessage = ({ data }) => {
+      assert(typeof data === 'object');
+      const {
+        type,
+        value
+      } = { ...data };
+      assert(typeof type === 'string');
+      switch (type) {
+        case 'pull':
+          if (this[kState].backpressurePromise !== undefined)
+            this[kState].backpressurePromise.resolve?.();
+          this[kState].backpressurePromise = undefined;
+          break;
+        case 'error':
+          writableStreamDefaultControllerErrorIfNeeded(
+            this[kState].controller,
+            value);
+          if (this[kState].backpressurePromise !== undefined)
+            this[kState].backpressurePromise.resolve?.();
+          this[kState].backpressurePromise = undefined;
+          break;
+      }
+    };
+    port.onmessageerror = () => {
+      const error = new CloneableDOMException(
+        'Internal transfered ReadableStream error',
+        'DataCloneError');
+      port.postMessage({ type: 'error', value: error });
+      writableStreamDefaultControllerErrorIfNeeded(
+        this[kState].controller,
+        error);
+      port.close();
+    };
+
+  }
+
+  start(controller) {
+    this[kState].controller = controller;
+  }
+
+  async write(chunk) {
+    if (this[kState].backpressurePromise === undefined) {
+      this[kState].backpressurePromise = {
+        promise: PromiseResolve(),
+        resolve: undefined,
+        reject: undefined,
+      };
+    }
+    await this[kState].backpressurePromise.promise;
+    this[kState].backpressurePromise = createDeferredPromise();
+    try {
+      this[kState].port.postMessage({ type: 'chunk', value: chunk });
+    } catch (error) {
+      if (error instanceof DOMException) {
+        // eslint-disable-next-line no-ex-assign
+        error = new CloneableDOMException(error.message, error.name);
+      }
+      this[kState].port.postMessage({ type: 'error', value: error });
+      this[kState].port.close();
+      throw error;
+    }
+  }
+
+  close() {
+    this[kState].port.postMessage({ type: 'close' });
+    this[kState].port.close();
+  }
+
+  abort(reason) {
+    try {
+      this[kState].port.postMessage({ type: 'error', value: reason });
+    } catch (error) {
+      if (error instanceof DOMException) {
+        // eslint-disable-next-line no-ex-assign
+        error = new CloneableDOMException(error.message, error.name);
+      }
+      this[kState].port.postMessage({ type: 'error', value: error });
+      throw error;
+    } finally {
+      this[kState].port.close();
+    }
+  }
+}
+
+function newCrossRealmReadableStream(writable, port) {
+  const readable =
+    new ReadableStream(
+      new CrossRealmTransformReadableSource(port));
+
+  const promise =
+    readableStreamPipeTo(readable, writable, false, false, false);
+
+  setPromiseHandled(promise);
+
+  return {
+    readable,
+    promise,
+  };
+}
+
+function newCrossRealmWritableSink(readable, port) {
+  const writable =
+    new WritableStream(
+      new CrossRealmTransformWritableSink(port));
+
+  const promise = readableStreamPipeTo(readable, writable, false, false, false);
+  setPromiseHandled(promise);
+  return {
+    writable,
+    promise,
+  };
+}
+
+module.exports = {
+  newCrossRealmReadableStream,
+  newCrossRealmWritableSink,
+  CrossRealmTransformWritableSink,
+  CrossRealmTransformReadableSource,
+  CloneableDOMException,
+  InternalCloneableDOMException,
+};

lib/internal/webstreams/transformstream.js

@@ -0,0 +1,591 @@
+'use strict';
+
+const {
+  FunctionPrototypeBind,
+  FunctionPrototypeCall,
+  ObjectDefineProperties,
+  PromisePrototypeCatch,
+  PromisePrototypeThen,
+  PromiseResolve,
+  ReflectConstruct,
+  SymbolToStringTag,
+} = primordials;
+
+const {
+  codes: {
+    ERR_ILLEGAL_CONSTRUCTOR,
+    ERR_INVALID_ARG_VALUE,
+    ERR_INVALID_STATE,
+    ERR_INVALID_THIS,
+  },
+} = require('internal/errors');
+
+const {
+  DOMException,
+} = internalBinding('messaging');
+
+const {
+  createDeferredPromise,
+  customInspectSymbol: kInspect,
+} = require('internal/util');
+
+const {
+  kDeserialize,
+  kTransfer,
+  kTransferList,
+  makeTransferable,
+} = require('internal/worker/js_transferable');
+
+const {
+  customInspect,
+  ensureIsPromise,
+  extractHighWaterMark,
+  extractSizeAlgorithm,
+  isBrandCheck,
+  nonOpFlush,
+  kType,
+  kState,
+} = require('internal/webstreams/util');
+
+const {
+  ReadableStream,
+  readableStreamDefaultControllerCanCloseOrEnqueue,
+  readableStreamDefaultControllerClose,
+  readableStreamDefaultControllerEnqueue,
+  readableStreamDefaultControllerError,
+  readableStreamDefaultControllerGetDesiredSize,
+  readableStreamDefaultControllerHasBackpressure,
+} = require('internal/webstreams/readablestream');
+
+const {
+  WritableStream,
+  writableStreamDefaultControllerErrorIfNeeded,
+} = require('internal/webstreams/writablestream');
+
+const assert = require('internal/assert');
+
+/**
+ * @typedef {import('./queuingstrategies').QueuingStrategy
+ * } QueuingStrategy
+ * @typedef {import('./queuingstrategies').QueuingStrategySize
+ * } QueuingStrategySize
+ *
+ * @callback TransformerStartCallback
+ * @param {TransformStreamDefaultController} controller;
+ *
+ * @callback TransformerFlushCallback
+ * @param {TransformStreamDefaultController} controller;
+ * @returns {Promise<void>}
+ *
+ * @callback TransformerTransformCallback
+ * @param {any} chunk
+ * @param {TransformStreamDefaultController} controller
+ * @returns {Promise<void>}
+ *
+ * @typedef {{
+ *  start? : TransformerStartCallback,
+ *  transform? : TransformerTransformCallback,
+ *  flush? : TransformerFlushCallback,
+ *  readableType? : any,
+ *  writableType? : any,
+ * }} Transformer
+ */
+
+class TransformStream {
+  [kType] = 'TransformStream';
+
+  get [SymbolToStringTag]() { return this[kType]; }
+
+  /**
+   * @param {Transformer} [transformer]
+   * @param {QueuingStrategy} [writableStrategy]
+   * @param {QueuingStrategy} [readableStrategy]
+   */
+  constructor(
+    transformer = null,
+    writableStrategy = {},
+    readableStrategy = {}) {
+    const readableType = transformer?.readableType;
+    const writableType = transformer?.writableType;
+    const start = transformer?.start;
+
+    if (readableType !== undefined) {
+      throw new ERR_INVALID_ARG_VALUE.RangeError(
+        'transformer.readableType',
+        readableType);
+    }
+    if (writableType !== undefined) {
+      throw new ERR_INVALID_ARG_VALUE.RangeError(
+        'transformer.writableType',
+        writableType);
+    }
+
+    const readableHighWaterMark = readableStrategy?.highWaterMark;
+    const readableSize = readableStrategy?.size;
+
+    const writableHighWaterMark = writableStrategy?.highWaterMark;
+    const writableSize = writableStrategy?.size;
+
+    const actualReadableHighWaterMark =
+      extractHighWaterMark(readableHighWaterMark, 0);
+    const actualReadableSize = extractSizeAlgorithm(readableSize);
+
+    const actualWritableHighWaterMark =
+      extractHighWaterMark(writableHighWaterMark, 1);
+    const actualWritableSize = extractSizeAlgorithm(writableSize);
+
+    const startPromise = createDeferredPromise();
+
+    initializeTransformStream(
+      this,
+      startPromise,
+      actualWritableHighWaterMark,
+      actualWritableSize,
+      actualReadableHighWaterMark,
+      actualReadableSize);
+
+    setupTransformStreamDefaultControllerFromTransformer(this, transformer);
+
+    if (start !== undefined) {
+      startPromise.resolve(
+        FunctionPrototypeCall(
+          start,
+          transformer,
+          this[kState].controller));
+    } else {
+      startPromise.resolve();
+    }
+
+    // eslint-disable-next-line no-constructor-return
+    return makeTransferable(this);
+  }
+
+  /**
+   * @readonly
+   * @type {ReadableStream}
+   */
+  get readable() {
+    if (!isTransformStream(this))
+      throw new ERR_INVALID_THIS('TransformStream');
+    return this[kState].readable;
+  }
+
+  /**
+   * @readonly
+   * @type {WritableStream}
+   */
+  get writable() {
+    if (!isTransformStream(this))
+      throw new ERR_INVALID_THIS('TransformStream');
+    return this[kState].writable;
+  }
+
+  [kInspect](depth, options) {
+    return customInspect(depth, options, this[kType], {
+      readable: this.readable,
+      writable: this.writable,
+      backpressure: this[kState].backpressure,
+    });
+  }
+
+  [kTransfer]() {
+    if (!isTransformStream(this))
+      throw new ERR_INVALID_THIS('TransformStream');
+    const {
+      readable,
+      writable,
+    } = this[kState];
+    if (readable.locked) {
+      throw new DOMException(
+        'Cannot transfer a locked ReadableStream',
+        'DataCloneError');
+    }
+    if (writable.locked) {
+      throw new DOMException(
+        'Cannot transfer a locked WritableStream',
+        'DataCloneError');
+    }
+    return {
+      data: {
+        readable,
+        writable,
+      },
+      deserializeInfo:
+        'internal/webstreams/transformstream:TransferedTransformStream'
+    };
+  }
+
+  [kTransferList]() {
+    return [ this[kState].readable, this[kState].writable ];
+  }
+
+  [kDeserialize]({ readable, writable }) {
+    this[kState].readable = readable;
+    this[kState].writable = writable;
+  }
+}
+
+ObjectDefineProperties(TransformStream.prototype, {
+  readable: { enumerable: true },
+  writable: { enumerable: true },
+});
+
+function TransferedTransformStream() {
+  return makeTransferable(ReflectConstruct(
+    function() {
+      this[kType] = 'TransformStream';
+      this[kState] = {
+        readable: undefined,
+        writable: undefined,
+        backpressure: undefined,
+        backpressureChange: {
+          promise: undefined,
+          resolve: undefined,
+          reject: undefined,
+        },
+        controller: undefined,
+      };
+    },
+    [], TransformStream));
+}
+TransferedTransformStream.prototype[kDeserialize] = () => {};
+
+class TransformStreamDefaultController {
+  [kType] = 'TransformStreamDefaultController';
+
+  get [SymbolToStringTag]() { return this[kType]; }
+
+  constructor() {
+    throw new ERR_ILLEGAL_CONSTRUCTOR();
+  }
+
+  /**
+   * @readonly
+   * @type {number}
+   */
+  get desiredSize() {
+    if (!isTransformStreamDefaultController(this))
+      throw new ERR_INVALID_THIS('TransformStreamDefaultController');
+    const {
+      stream,
+    } = this[kState];
+    const {
+      readable,
+    } = stream[kState];
+    const {
+      controller: readableController,
+    } = readable[kState];
+    return readableStreamDefaultControllerGetDesiredSize(readableController);
+  }
+
+  /**
+   * @param {any} chunk
+   */
+  enqueue(chunk = undefined) {
+    if (!isTransformStreamDefaultController(this))
+      throw new ERR_INVALID_THIS('TransformStreamDefaultController');
+    transformStreamDefaultControllerEnqueue(this, chunk);
+  }
+
+  /**
+   * @param {any} reason
+   */
+  error(reason = undefined) {
+    if (!isTransformStreamDefaultController(this))
+      throw new ERR_INVALID_THIS('TransformStreamDefaultController');
+    transformStreamDefaultControllerError(this, reason);
+  }
+
+  terminate() {
+    if (!isTransformStreamDefaultController(this))
+      throw new ERR_INVALID_THIS('TransformStreamDefaultController');
+    transformStreamDefaultControllerTerminate(this);
+  }
+
+  [kInspect](depth, options) {
+    return customInspect(depth, options, this[kType], {
+      stream: this[kState].stream,
+    });
+  }
+}
+
+ObjectDefineProperties(TransformStreamDefaultController.prototype, {
+  desiredSize: { enumerable: true },
+  enqueue: { enumerable: true },
+  error: { enumerable: true },
+  terminate: { enumerable: true },
+});
+
+function createTransformStreamDefaultController() {
+  return ReflectConstruct(
+    function() {
+      this[kType] = 'TransformStreamDefaultController';
+    },
+    [],
+    TransformStreamDefaultController);
+}
+
+const isTransformStream =
+  isBrandCheck('TransformStream');
+const isTransformStreamDefaultController =
+  isBrandCheck('TransformStreamDefaultController');
+
+async function defaultTransformAlgorithm(chunk, controller) {
+  transformStreamDefaultControllerEnqueue(controller, chunk);
+}
+
+function initializeTransformStream(
+  stream,
+  startPromise,
+  writableHighWaterMark,
+  writableSizeAlgorithm,
+  readableHighWaterMark,
+  readableSizeAlgorithm) {
+
+  const writable = new WritableStream({
+    start() { return startPromise.promise; },
+    write(chunk) {
+      return transformStreamDefaultSinkWriteAlgorithm(stream, chunk);
+    },
+    abort(reason) {
+      return transformStreamDefaultSinkAbortAlgorithm(stream, reason);
+    },
+    close() {
+      return transformStreamDefaultSinkCloseAlgorithm(stream);
+    },
+  }, {
+    highWaterMark: writableHighWaterMark,
+    size: writableSizeAlgorithm,
+  });
+
+  const readable = new ReadableStream({
+    start() { return startPromise.promise; },
+    pull() {
+      return transformStreamDefaultSourcePullAlgorithm(stream);
+    },
+    cancel(reason) {
+      transformStreamErrorWritableAndUnblockWrite(stream, reason);
+      return PromiseResolve();
+    },
+  }, {
+    highWaterMark: readableHighWaterMark,
+    size: readableSizeAlgorithm,
+  });
+
+  stream[kState] = {
+    readable,
+    writable,
+    controller: undefined,
+    backpressure: undefined,
+    backpressureChange: {
+      promise: undefined,
+      resolve: undefined,
+      reject: undefined,
+    }
+  };
+
+  transformStreamSetBackpressure(stream, true);
+}
+
+function transformStreamError(stream, error) {
+  const {
+    readable,
+  } = stream[kState];
+  const {
+    controller,
+  } = readable[kState];
+  readableStreamDefaultControllerError(controller, error);
+  transformStreamErrorWritableAndUnblockWrite(stream, error);
+}
+
+function transformStreamErrorWritableAndUnblockWrite(stream, error) {
+  const {
+    controller,
+    writable,
+  } = stream[kState];
+  transformStreamDefaultControllerClearAlgorithms(controller);
+  writableStreamDefaultControllerErrorIfNeeded(
+    writable[kState].controller,
+    error);
+  if (stream[kState].backpressure)
+    transformStreamSetBackpressure(stream, false);
+}
+
+function transformStreamSetBackpressure(stream, backpressure) {
+  assert(stream[kState].backpressure !== backpressure);
+  if (stream[kState].backpressureChange.promise !== undefined)
+    stream[kState].backpressureChange.resolve?.();
+  stream[kState].backpressureChange = createDeferredPromise();
+  stream[kState].backpressure = backpressure;
+}
+
+function setupTransformStreamDefaultController(
+  stream,
+  controller,
+  transformAlgorithm,
+  flushAlgorithm) {
+  assert(isTransformStream(stream));
+  assert(stream[kState].controller === undefined);
+  controller[kState] = {
+    stream,
+    transformAlgorithm,
+    flushAlgorithm,
+  };
+  stream[kState].controller = controller;
+}
+
+function setupTransformStreamDefaultControllerFromTransformer(
+  stream,
+  transformer) {
+  const controller = createTransformStreamDefaultController();
+  const transform = transformer?.transform || defaultTransformAlgorithm;
+  const flush = transformer?.flush || nonOpFlush;
+  const transformAlgorithm =
+    FunctionPrototypeBind(transform, transformer);
+  const flushAlgorithm =
+    FunctionPrototypeBind(flush, transformer);
+
+  setupTransformStreamDefaultController(
+    stream,
+    controller,
+    transformAlgorithm,
+    flushAlgorithm);
+}
+
+function transformStreamDefaultControllerClearAlgorithms(controller) {
+  controller[kState].transformAlgorithm = undefined;
+  controller[kState].flushAlgorithm = undefined;
+}
+
+function transformStreamDefaultControllerEnqueue(controller, chunk) {
+  const {
+    stream,
+  } = controller[kState];
+  const {
+    readable,
+  } = stream[kState];
+  const {
+    controller: readableController,
+  } = readable[kState];
+  if (!readableStreamDefaultControllerCanCloseOrEnqueue(readableController))
+    throw new ERR_INVALID_STATE.TypeError('Unable to enqueue');
+  try {
+    readableStreamDefaultControllerEnqueue(readableController, chunk);
+  } catch (error) {
+    transformStreamErrorWritableAndUnblockWrite(stream, error);
+    throw readable[kState].storedError;
+  }
+  const backpressure =
+    readableStreamDefaultControllerHasBackpressure(readableController);
+  if (backpressure !== stream[kState].backpressure) {
+    assert(backpressure);
+    transformStreamSetBackpressure(stream, true);
+  }
+}
+
+function transformStreamDefaultControllerError(controller, error) {
+  transformStreamError(controller[kState].stream, error);
+}
+
+function transformStreamDefaultControllerPerformTransform(controller, chunk) {
+  const transformPromise =
+    ensureIsPromise(
+      controller[kState].transformAlgorithm,
+      controller,
+      chunk,
+      controller);
+  return PromisePrototypeCatch(
+    transformPromise,
+    (error) => {
+      transformStreamError(controller[kState].stream, error);
+      throw error;
+    });
+}
+
+function transformStreamDefaultControllerTerminate(controller) {
+  const {
+    stream,
+  } = controller[kState];
+  const {
+    readable,
+  } = stream[kState];
+  assert(readable !== undefined);
+  const {
+    controller: readableController,
+  } = readable[kState];
+  readableStreamDefaultControllerClose(readableController);
+  transformStreamErrorWritableAndUnblockWrite(
+    stream,
+    new ERR_INVALID_STATE.TypeError('TransformStream has been terminated'));
+}
+
+function transformStreamDefaultSinkWriteAlgorithm(stream, chunk) {
+  const {
+    writable,
+    controller,
+  } = stream[kState];
+  assert(writable[kState].state === 'writable');
+  if (stream[kState].backpressure) {
+    const backpressureChange = stream[kState].backpressureChange.promise;
+    return PromisePrototypeThen(
+      backpressureChange,
+      () => {
+        const {
+          writable,
+        } = stream[kState];
+        if (writable[kState].state === 'erroring')
+          throw writable[kState].storedError;
+        assert(writable[kState].state === 'writable');
+        return transformStreamDefaultControllerPerformTransform(
+          controller,
+          chunk);
+      });
+  }
+  return transformStreamDefaultControllerPerformTransform(controller, chunk);
+}
+
+async function transformStreamDefaultSinkAbortAlgorithm(stream, reason) {
+  transformStreamError(stream, reason);
+}
+
+function transformStreamDefaultSinkCloseAlgorithm(stream) {
+  const {
+    readable,
+    controller,
+  } = stream[kState];
+
+  const flushPromise =
+    ensureIsPromise(
+      controller[kState].flushAlgorithm,
+      controller,
+      controller);
+  transformStreamDefaultControllerClearAlgorithms(controller);
+  return PromisePrototypeThen(
+    flushPromise,
+    () => {
+      if (readable[kState].state === 'errored')
+        throw readable[kState].storedError;
+      readableStreamDefaultControllerClose(readable[kState].controller);
+    },
+    (error) => {
+      transformStreamError(stream, error);
+      throw readable[kState].storedError;
+    });
+}
+
+function transformStreamDefaultSourcePullAlgorithm(stream) {
+  assert(stream[kState].backpressure);
+  assert(stream[kState].backpressureChange.promise !== undefined);
+  transformStreamSetBackpressure(stream, false);
+  return stream[kState].backpressureChange.promise;
+}
+
+module.exports = {
+  TransformStream,
+  TransformStreamDefaultController,
+  TransferedTransformStream,
+
+  // Exported Brand Checks
+  isTransformStream,
+  isTransformStreamDefaultController,
+};

lib/internal/webstreams/util.js

@@ -0,0 +1,237 @@
+'use strict';
+
+const {
+  ArrayBufferPrototype,
+  ArrayPrototypePush,
+  ArrayPrototypeShift,
+  AsyncIteratorPrototype,
+  FunctionPrototypeCall,
+  MathMax,
+  NumberIsNaN,
+  ObjectCreate,
+  PromisePrototypeThen,
+  PromiseResolve,
+  PromiseReject,
+  ReflectGet,
+  Symbol,
+} = primordials;
+
+const {
+  codes: {
+    ERR_INVALID_ARG_TYPE,
+    ERR_INVALID_ARG_VALUE,
+    ERR_OPERATION_FAILED,
+  },
+} = require('internal/errors');
+
+const {
+  copyArrayBuffer,
+  detachArrayBuffer
+} = internalBinding('buffer');
+
+const {
+  isPromise,
+} = require('util/types');
+
+const {
+  inspect,
+} = require('util');
+
+const {
+  getPromiseDetails,
+  kPending,
+} = internalBinding('util');
+
+const assert = require('internal/assert');
+
+const kState = Symbol('kState');
+const kType = Symbol('kType');
+
+const AsyncIterator = ObjectCreate(AsyncIteratorPrototype, {
+  next: {
+    configurable: true,
+    enumerable: true,
+    writable: true,
+  },
+  return: {
+    configurable: true,
+    enumerable: true,
+    writable: true,
+  },
+});
+
+function extractHighWaterMark(value, defaultHWM) {
+  if (value === undefined) return defaultHWM;
+  value = +value;
+  if (typeof value !== 'number' ||
+      NumberIsNaN(value) ||
+      value < 0)
+    throw new ERR_INVALID_ARG_VALUE.RangeError('strategy.highWaterMark', value);
+  return value;
+}
+
+function extractSizeAlgorithm(size) {
+  if (size === undefined) return () => 1;
+  if (typeof size !== 'function')
+    throw new ERR_INVALID_ARG_TYPE('strategy.size', 'Function', size);
+  return size;
+}
+
+function customInspect(depth, options, name, data) {
+  if (depth < 0)
+    return this;
+
+  const opts = {
+    ...options,
+    depth: options.depth == null ? null : options.depth - 1
+  };
+
+  return `${name} ${inspect(data, opts)}`;
+}
+
+// These are defensive to work around the possibility that
+// the buffer, byteLength, and byteOffset properties on
+// ArrayBuffer and ArrayBufferView's may have been tampered with.
+
+function ArrayBufferViewGetBuffer(view) {
+  return ReflectGet(view.constructor.prototype, 'buffer', view);
+}
+
+function ArrayBufferViewGetByteLength(view) {
+  return ReflectGet(view.constructor.prototype, 'byteLength', view);
+}
+
+function ArrayBufferViewGetByteOffset(view) {
+  return ReflectGet(view.constructor.prototype, 'byteOffset', view);
+}
+
+function ArrayBufferGetByteLength(view) {
+  return ReflectGet(ArrayBufferPrototype, 'byteLength', view);
+}
+
+function isBrandCheck(brand) {
+  return (value) => {
+    return value != null &&
+           value[kState] !== undefined &&
+           value[kType] === brand;
+  };
+}
+
+function transferArrayBuffer(buffer) {
+  const res = detachArrayBuffer(buffer);
+  if (res === undefined) {
+    throw new ERR_OPERATION_FAILED.TypeError(
+      'The ArrayBuffer could not be transfered');
+  }
+  return res;
+}
+
+function dequeueValue(controller) {
+  assert(controller[kState].queue !== undefined);
+  assert(controller[kState].queueTotalSize !== undefined);
+  assert(controller[kState].queue.length);
+  const {
+    value,
+    size,
+  } = ArrayPrototypeShift(controller[kState].queue);
+  controller[kState].queueTotalSize =
+    MathMax(0, controller[kState].queueTotalSize - size);
+  return value;
+}
+
+function resetQueue(controller) {
+  assert(controller[kState].queue !== undefined);
+  assert(controller[kState].queueTotalSize !== undefined);
+  controller[kState].queue = [];
+  controller[kState].queueTotalSize = 0;
+}
+
+function peekQueueValue(controller) {
+  assert(controller[kState].queue !== undefined);
+  assert(controller[kState].queueTotalSize !== undefined);
+  assert(controller[kState].queue.length);
+  return controller[kState].queue[0].value;
+}
+
+function enqueueValueWithSize(controller, value, size) {
+  assert(controller[kState].queue !== undefined);
+  assert(controller[kState].queueTotalSize !== undefined);
+  size = +size;
+  if (typeof size !== 'number' ||
+      size < 0 ||
+      NumberIsNaN(size) ||
+      size === Infinity) {
+    throw new ERR_INVALID_ARG_VALUE.RangeError('size', size);
+  }
+  ArrayPrototypePush(controller[kState].queue, { value, size });
+  controller[kState].queueTotalSize += size;
+}
+
+function ensureIsPromise(fn, thisArg, ...args) {
+  try {
+    const value = FunctionPrototypeCall(fn, thisArg, ...args);
+    return isPromise(value) ? value : PromiseResolve(value);
+  } catch (error) {
+    return PromiseReject(error);
+  }
+}
+
+function isPromisePending(promise) {
+  if (promise === undefined) return false;
+  const details = getPromiseDetails(promise);
+  return details?.[0] === kPending;
+}
+
+function setPromiseHandled(promise) {
+  // Alternatively, we could use the native API
+  // MarkAsHandled, but this avoids the extra boundary cross
+  // and is hopefully faster at the cost of an extra Promise
+  // allocation.
+  PromisePrototypeThen(promise, () => {}, () => {});
+}
+
+async function nonOpFlush() {}
+
+function nonOpStart() {}
+
+async function nonOpPull() {}
+
+async function nonOpCancel() {}
+
+async function nonOpWrite() {}
+
+let transfer;
+function lazyTransfer() {
+  if (transfer === undefined)
+    transfer = require('internal/webstreams/transfer');
+  return transfer;
+}
+
+module.exports = {
+  ArrayBufferViewGetBuffer,
+  ArrayBufferViewGetByteLength,
+  ArrayBufferViewGetByteOffset,
+  ArrayBufferGetByteLength,
+  AsyncIterator,
+  copyArrayBuffer,
+  customInspect,
+  dequeueValue,
+  ensureIsPromise,
+  enqueueValueWithSize,
+  extractHighWaterMark,
+  extractSizeAlgorithm,
+  lazyTransfer,
+  isBrandCheck,
+  isPromisePending,
+  peekQueueValue,
+  resetQueue,
+  setPromiseHandled,
+  transferArrayBuffer,
+  nonOpCancel,
+  nonOpFlush,
+  nonOpPull,
+  nonOpStart,
+  nonOpWrite,
+  kType,
+  kState,
+};

lib/stream/web.js

@@ -0,0 +1,48 @@
+'use strict';
+
+const {
+  emitExperimentalWarning,
+} = require('internal/util');
+
+emitExperimentalWarning('stream/web');
+
+const {
+  TransformStream,
+  TransformStreamDefaultController,
+} = require('internal/webstreams/transformstream');
+
+const {
+  WritableStream,
+  WritableStreamDefaultController,
+  WritableStreamDefaultWriter,
+} = require('internal/webstreams/writablestream');
+
+const {
+  ReadableStream,
+  ReadableStreamDefaultReader,
+  ReadableStreamBYOBReader,
+  ReadableStreamBYOBRequest,
+  ReadableByteStreamController,
+  ReadableStreamDefaultController,
+} = require('internal/webstreams/readablestream');
+
+const {
+  ByteLengthQueuingStrategy,
+  CountQueuingStrategy,
+} = require('internal/webstreams/queuingstrategies');
+
+module.exports = {
+  ReadableStream,
+  ReadableStreamDefaultReader,
+  ReadableStreamBYOBReader,
+  ReadableStreamBYOBRequest,
+  ReadableByteStreamController,
+  ReadableStreamDefaultController,
+  TransformStream,
+  TransformStreamDefaultController,
+  WritableStream,
+  WritableStreamDefaultWriter,
+  WritableStreamDefaultController,
+  ByteLengthQueuingStrategy,
+  CountQueuingStrategy,
+};

src/node_buffer.cc

@@ -67,6 +67,7 @@ using v8::MaybeLocal;
 using v8::Nothing;
 using v8::Number;
 using v8::Object;
+using v8::SharedArrayBuffer;
 using v8::String;
 using v8::Uint32;
 using v8::Uint32Array;
@@ -1158,6 +1159,60 @@ void GetZeroFillToggle(const FunctionCallbackInfo<Value>& args) {
   args.GetReturnValue().Set(Uint32Array::New(ab, 0, 1));
 }
 
+void DetachArrayBuffer(const FunctionCallbackInfo<Value>& args) {
+  Environment* env = Environment::GetCurrent(args);
+  if (args[0]->IsArrayBuffer()) {
+    Local<ArrayBuffer> buf = args[0].As<ArrayBuffer>();
+    if (buf->IsDetachable()) {
+      std::shared_ptr<BackingStore> store = buf->GetBackingStore();
+      buf->Detach();
+      args.GetReturnValue().Set(ArrayBuffer::New(env->isolate(), store));
+    }
+  }
+}
+
+void CopyArrayBuffer(const FunctionCallbackInfo<Value>& args) {
+  // args[0] == Destination ArrayBuffer
+  // args[1] == Destination ArrayBuffer Offset
+  // args[2] == Source ArrayBuffer
+  // args[3] == Source ArrayBuffer Offset
+  // args[4] == bytesToCopy
+
+  CHECK(args[0]->IsArrayBuffer() || args[0]->IsSharedArrayBuffer());
+  CHECK(args[1]->IsUint32());
+  CHECK(args[2]->IsArrayBuffer() || args[2]->IsSharedArrayBuffer());
+  CHECK(args[3]->IsUint32());
+  CHECK(args[4]->IsUint32());
+
+  std::shared_ptr<BackingStore> destination;
+  std::shared_ptr<BackingStore> source;
+
+  if (args[0]->IsArrayBuffer()) {
+    destination = args[0].As<ArrayBuffer>()->GetBackingStore();
+  } else if (args[0]->IsSharedArrayBuffer()) {
+    destination = args[0].As<SharedArrayBuffer>()->GetBackingStore();
+  }
+
+  if (args[2]->IsArrayBuffer()) {
+    source = args[2].As<ArrayBuffer>()->GetBackingStore();
+  } else if (args[0]->IsSharedArrayBuffer()) {
+    source = args[2].As<SharedArrayBuffer>()->GetBackingStore();
+  }
+
+  uint32_t destination_offset = args[1].As<Uint32>()->Value();
+  uint32_t source_offset = args[3].As<Uint32>()->Value();
+  size_t bytes_to_copy = args[4].As<Uint32>()->Value();
+
+  CHECK_GE(destination->ByteLength() - destination_offset, bytes_to_copy);
+  CHECK_GE(source->ByteLength() - source_offset, bytes_to_copy);
+
+  uint8_t* dest =
+      static_cast<uint8_t*>(destination->Data()) + destination_offset;
+  uint8_t* src =
+      static_cast<uint8_t*>(source->Data()) + source_offset;
+  memcpy(dest, src, bytes_to_copy);
+}
+
 void Initialize(Local<Object> target,
                 Local<Value> unused,
                 Local<Context> context,
@@ -1176,6 +1231,9 @@ void Initialize(Local<Object> target,
   env->SetMethodNoSideEffect(target, "indexOfNumber", IndexOfNumber);
   env->SetMethodNoSideEffect(target, "indexOfString", IndexOfString);
 
+  env->SetMethod(target, "detachArrayBuffer", DetachArrayBuffer);
+  env->SetMethod(target, "copyArrayBuffer", CopyArrayBuffer);
+
   env->SetMethod(target, "swap16", Swap16);
   env->SetMethod(target, "swap32", Swap32);
   env->SetMethod(target, "swap64", Swap64);
@@ -1251,6 +1309,9 @@ void RegisterExternalReferences(ExternalReferenceRegistry* registry) {
   registry->Register(StringWrite<UTF8>);
   registry->Register(GetZeroFillToggle);
 
+  registry->Register(DetachArrayBuffer);
+  registry->Register(CopyArrayBuffer);
+
   Blob::RegisterExternalReferences(registry);
   FixedSizeBlobCopyJob::RegisterExternalReferences(registry);
 }

test/fixtures/wpt/README.md

@@ -21,7 +21,7 @@ Last update:
 - html/webappapis/timers: https://github.com/web-platform-tests/wpt/tree/5873f2d8f1/html/webappapis/timers
 - interfaces: https://github.com/web-platform-tests/wpt/tree/fcb671ed8b/interfaces
 - resources: https://github.com/web-platform-tests/wpt/tree/972ca5b669/resources
-- streams: https://github.com/web-platform-tests/wpt/tree/b869e60df1/streams
+- streams: https://github.com/web-platform-tests/wpt/tree/8f60d94439/streams
 - url: https://github.com/web-platform-tests/wpt/tree/1fcb39223d/url
 
 [Web Platform Tests]: https://github.com/web-platform-tests/wpt

test/fixtures/wpt/streams/readable-byte-streams/general.any.js

@@ -957,7 +957,8 @@ promise_test(() => {
     assert_equals(view.byteOffset, 0, 'byteOffset');
     assert_equals(view.byteLength, 2, 'byteLength');
 
-    assert_equals(view[0], 0x0201);
+    const dataView = new DataView(view.buffer, view.byteOffset, view.byteLength);
+    assert_equals(dataView.getUint16(0), 0x0102);
 
     return reader.read(new Uint8Array(1));
   }).then(result => {
@@ -1138,7 +1139,7 @@ promise_test(() => {
 
     assert_equals(pullCount, 1, '1 pull() should have been made in response to partial fill by enqueue()');
     assert_not_equals(byobRequest, null, 'byobRequest should not be null');
-    assert_equals(viewInfos[0].byteLength, 2, 'byteLength before enqueue() shouild be 2');
+    assert_equals(viewInfos[0].byteLength, 2, 'byteLength before enqueue() should be 2');
     assert_equals(viewInfos[1].byteLength, 1, 'byteLength after enqueue() should be 1');
 
     reader.cancel();
@@ -1326,7 +1327,9 @@ promise_test(() => {
     const view = result.value;
     assert_equals(view.byteOffset, 0);
     assert_equals(view.byteLength, 2);
-    assert_equals(view[0], 0xaaff);
+
+    const dataView = new DataView(view.buffer, view.byteOffset, view.byteLength);
+    assert_equals(dataView.getUint16(0), 0xffaa);
 
     assert_equals(viewInfo.constructor, Uint8Array, 'view.constructor should be Uint8Array');
     assert_equals(viewInfo.bufferByteLength, 2, 'view.buffer.byteLength should be 2');
@@ -1381,7 +1384,9 @@ promise_test(() => {
       assert_equals(view.buffer.byteLength, 4, 'buffer.byteLength');
       assert_equals(view.byteOffset, 0, 'byteOffset');
       assert_equals(view.byteLength, 2, 'byteLength');
-      assert_equals(view[0], 0x0001, 'Contents are set');
+
+      const dataView = new DataView(view.buffer, view.byteOffset, view.byteLength);
+      assert_equals(dataView.getUint16(0), 0x0100, 'contents are set');
 
       const p = reader.read(new Uint16Array(1));
 
@@ -1395,7 +1400,9 @@ promise_test(() => {
       assert_equals(view.buffer.byteLength, 2, 'buffer.byteLength');
       assert_equals(view.byteOffset, 0, 'byteOffset');
       assert_equals(view.byteLength, 2, 'byteLength');
-      assert_equals(view[0], 0x0302, 'Contents are set');
+
+      const dataView = new DataView(view.buffer, view.byteOffset, view.byteLength);
+      assert_equals(dataView.getUint16(0), 0x0203, 'contents are set');
 
       assert_not_equals(byobRequest, null, 'byobRequest must not be null');
       assert_equals(viewInfo.constructor, Uint8Array, 'view.constructor should be Uint8Array');

test/fixtures/wpt/streams/readable-byte-streams/non-transferable-buffers.any.js

@@ -25,7 +25,7 @@ test(t => {
 
   const memory = new WebAssembly.Memory({ initial: 1 });
   const view = new Uint8Array(memory.buffer, 0, 1);
-  assert_throws_js(t, TypeError, controller.enqueue(view));
+  assert_throws_js(TypeError, () => controller.enqueue(view));
 }, 'ReadableStream with byte source: enqueue() with a non-transferable buffer');
 
 promise_test(async t => {
@@ -54,5 +54,5 @@ promise_test(async t => {
   );
   await pullCalledPromise;
 
-  assert_throws_js(t, TypeError, byobRequest.respondWithNewView(newView));
+  assert_throws_js(TypeError, () => byobRequest.respondWithNewView(newView));
 }, 'ReadableStream with byte source: respondWithNewView() with a non-transferable buffer');

test/fixtures/wpt/streams/writable-streams/aborting.any.js

@@ -1376,3 +1376,111 @@ promise_test(t => {
   return promise_rejects_js(t, TypeError, ws.abort(), 'abort should reject')
     .then(() => writer.ready);
 }, 'abort on a locked stream should reject');
+
+test(t => {
+  let ctrl;
+  const ws = new WritableStream({start(c) { ctrl = c; }});
+  const e = Error('hello');
+
+  assert_true(ctrl.signal instanceof AbortSignal);
+  assert_false(ctrl.signal.aborted);
+  assert_equals(ctrl.abortReason, undefined);
+  ws.abort(e);
+  assert_true(ctrl.signal.aborted);
+  assert_equals(ctrl.abortReason, e);
+}, 'WritableStreamDefaultController.signal');
+
+promise_test(async t => {
+  let ctrl;
+  let resolve;
+  const called = new Promise(r => resolve = r);
+
+  const ws = new WritableStream({
+    start(c) { ctrl = c; },
+    write() { resolve(); return new Promise(() => {}); }
+  });
+  const writer = ws.getWriter();
+
+  writer.write(99);
+  await called;
+
+  assert_false(ctrl.signal.aborted);
+  assert_equals(ctrl.abortReason, undefined);
+  writer.abort();
+  assert_true(ctrl.signal.aborted);
+  assert_equals(ctrl.abortReason, undefined);
+}, 'the abort signal is signalled synchronously - write');
+
+promise_test(async t => {
+  let ctrl;
+  let resolve;
+  const called = new Promise(r => resolve = r);
+
+  const ws = new WritableStream({
+    start(c) { ctrl = c; },
+    close() { resolve(); return new Promise(() => {}); }
+  });
+  const writer = ws.getWriter();
+
+  writer.close(99);
+  await called;
+
+  assert_false(ctrl.signal.aborted);
+  writer.abort();
+  assert_true(ctrl.signal.aborted);
+}, 'the abort signal is signalled synchronously - close');
+
+promise_test(async t => {
+  let ctrl;
+  const ws = new WritableStream({start(c) { ctrl = c; }});
+  const writer = ws.getWriter();
+
+  const e = TypeError();
+  ctrl.error(e);
+  await promise_rejects_exactly(t, e, writer.closed);
+  assert_false(ctrl.signal.aborted);
+}, 'the abort signal is not signalled on error');
+
+promise_test(async t => {
+  let ctrl;
+  const e = TypeError();
+  const ws = new WritableStream({
+    start(c) { ctrl = c; },
+    async write() { throw e; }
+  });
+  const writer = ws.getWriter();
+
+  await promise_rejects_exactly(t, e, writer.write('hello'), 'write result');
+  await promise_rejects_exactly(t, e, writer.closed, 'closed');
+  assert_false(ctrl.signal.aborted);
+}, 'the abort signal is not signalled on write failure');
+
+promise_test(async t => {
+  let ctrl;
+  const e = TypeError();
+  const ws = new WritableStream({
+    start(c) { ctrl = c; },
+    async close() { throw e; }
+  });
+  const writer = ws.getWriter();
+
+  await promise_rejects_exactly(t, e, writer.close(), 'close result');
+  await promise_rejects_exactly(t, e, writer.closed, 'closed');
+  assert_false(ctrl.signal.aborted);
+}, 'the abort signal is not signalled on close failure');
+
+promise_test(async t => {
+  let ctrl;
+  const e1 = SyntaxError();
+  const e2 = TypeError();
+  const ws = new WritableStream({
+    start(c) { ctrl = c; },
+  });
+
+  const writer = ws.getWriter();
+  ctrl.signal.addEventListener('abort', () => writer.abort(e2));
+  writer.abort(e1);
+  assert_true(ctrl.signal.aborted);
+
+  await promise_rejects_exactly(t, e2, writer.closed, 'closed');
+}, 'recursive abort() call');

test/fixtures/wpt/versions.json

@@ -44,7 +44,7 @@
     "path": "resources"
   },
   "streams": {
-    "commit": "b869e60df1b8d3840e09b41c5e987c7e23f6856c",
+    "commit": "8f60d9443949c323522a2009518d54d5d6ab5541",
     "path": "streams"
   },
   "url": {

test/parallel/test-whatwg-readablebytestream.js

@@ -0,0 +1,238 @@
+// Flags: --expose-internals --no-warnings
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+
+const {
+  ReadableStream,
+  ReadableByteStreamController,
+  ReadableStreamDefaultReader,
+  ReadableStreamBYOBReader,
+  ReadableStreamBYOBRequest,
+} = require('stream/web');
+
+const {
+  kState,
+} = require('internal/webstreams/util');
+
+const {
+  open,
+} = require('fs/promises');
+
+const {
+  readFileSync,
+} = require('fs');
+
+const {
+  Buffer,
+} = require('buffer');
+
+const {
+  inspect,
+} = require('util');
+
+{
+  const r = new ReadableStream({
+    type: 'bytes',
+  });
+
+  assert(r[kState].controller instanceof ReadableByteStreamController);
+
+  assert.strictEqual(typeof r.locked, 'boolean');
+  assert.strictEqual(typeof r.cancel, 'function');
+  assert.strictEqual(typeof r.getReader, 'function');
+  assert.strictEqual(typeof r.pipeThrough, 'function');
+  assert.strictEqual(typeof r.pipeTo, 'function');
+  assert.strictEqual(typeof r.tee, 'function');
+
+  ['', null, 'asdf'].forEach((mode) => {
+    assert.throws(() => r.getReader({ mode }), {
+      code: 'ERR_INVALID_ARG_VALUE',
+    });
+  });
+
+  [1, 'asdf'].forEach((options) => {
+    assert.throws(() => r.getReader(options), {
+      code: 'ERR_INVALID_ARG_TYPE',
+    });
+  });
+
+  assert(!r.locked);
+  const defaultReader = r.getReader();
+  assert(r.locked);
+  assert(defaultReader instanceof ReadableStreamDefaultReader);
+  defaultReader.releaseLock();
+  const byobReader = r.getReader({ mode: 'byob' });
+  assert(byobReader instanceof ReadableStreamBYOBReader);
+}
+
+class Source {
+  constructor() {
+    this.controllerClosed = false;
+  }
+
+  async start(controller) {
+    this.file = await open(__filename);
+    this.controller = controller;
+  }
+
+  async pull(controller) {
+    const byobRequest = controller.byobRequest;
+    assert.match(inspect(byobRequest), /ReadableStreamBYOBRequest/);
+
+    const view = byobRequest.view;
+    const {
+      bytesRead,
+    } = await this.file.read({
+      buffer: view,
+      offset: view.byteOffset,
+      length: view.byteLength
+    });
+
+    if (bytesRead === 0) {
+      await this.file.close();
+      this.controller.close();
+    }
+
+    assert.throws(() => byobRequest.respondWithNewView({}), {
+      code: 'ERR_INVALID_ARG_TYPE',
+    });
+
+    byobRequest.respond(bytesRead);
+
+    assert.throws(() => byobRequest.respond(bytesRead), {
+      code: 'ERR_INVALID_STATE',
+    });
+    assert.throws(() => byobRequest.respondWithNewView(view), {
+      code: 'ERR_INVALID_STATE',
+    });
+  }
+
+  get type() { return 'bytes'; }
+
+  get autoAllocateChunkSize() { return 1024; }
+}
+
+{
+  const stream = new ReadableStream(new Source());
+  assert(stream[kState].controller instanceof ReadableByteStreamController);
+
+  async function read(stream) {
+    const reader = stream.getReader({ mode: 'byob' });
+
+    const chunks = [];
+    let result;
+    do {
+      result = await reader.read(Buffer.alloc(100));
+      if (result.value !== undefined)
+        chunks.push(Buffer.from(result.value));
+    } while (!result.done);
+
+    return Buffer.concat(chunks);
+  }
+
+  read(stream).then(common.mustCall((data) => {
+    const check = readFileSync(__filename);
+    assert.deepStrictEqual(check, data);
+  }));
+}
+
+{
+  const stream = new ReadableStream(new Source());
+  assert(stream[kState].controller instanceof ReadableByteStreamController);
+
+  async function read(stream) {
+    const chunks = [];
+    for await (const chunk of stream)
+      chunks.push(chunk);
+
+    return Buffer.concat(chunks);
+  }
+
+  read(stream).then(common.mustCall((data) => {
+    const check = readFileSync(__filename);
+    assert.deepStrictEqual(check, data);
+  }));
+}
+
+{
+  const stream = new ReadableStream(new Source());
+  assert(stream[kState].controller instanceof ReadableByteStreamController);
+
+  async function read(stream) {
+    // eslint-disable-next-line no-unused-vars
+    for await (const _ of stream)
+      break;
+  }
+
+  read(stream).then(common.mustCall());
+}
+
+{
+  const stream = new ReadableStream(new Source());
+  assert(stream[kState].controller instanceof ReadableByteStreamController);
+
+  const error = new Error('boom');
+
+  async function read(stream) {
+    // eslint-disable-next-line no-unused-vars
+    for await (const _ of stream)
+      throw error;
+  }
+
+  assert.rejects(read(stream), error);
+}
+
+{
+  assert.throws(() => {
+    Reflect.get(ReadableStreamBYOBRequest.prototype, 'view', {});
+  }, {
+    code: 'ERR_INVALID_THIS',
+  });
+
+  assert.throws(() => ReadableStreamBYOBRequest.prototype.respond.call({}), {
+    code: 'ERR_INVALID_THIS',
+  });
+
+  assert.throws(() => {
+    ReadableStreamBYOBRequest.prototype.respondWithNewView.call({});
+  }, {
+    code: 'ERR_INVALID_THIS',
+  });
+}
+
+{
+  const readable = new ReadableStream({ type: 'bytes' });
+  const reader = readable.getReader({ mode: 'byob' });
+  reader.releaseLock();
+  reader.releaseLock();
+  assert.rejects(reader.read(new Uint8Array(10)), {
+    code: 'ERR_INVALID_STATE',
+  });
+  assert.rejects(reader.cancel(), {
+    code: 'ERR_INVALID_STATE',
+  });
+}
+
+{
+  let controller;
+  new ReadableStream({
+    type: 'bytes',
+    start(c) { controller = c; }
+  });
+  assert.throws(() => controller.enqueue(1), {
+    code: 'ERR_INVALID_ARG_TYPE',
+  });
+  controller.close();
+  assert.throws(() => controller.enqueue(new Uint8Array(10)), {
+    code: 'ERR_INVALID_STATE',
+  });
+  assert.throws(() => controller.close(), {
+    code: 'ERR_INVALID_STATE',
+  });
+}
+
+{
+
+}

test/parallel/test-whatwg-transformstream.js

@@ -0,0 +1,188 @@
+// Flags: --expose-internals --no-warnings
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+
+const {
+  ReadableStream,
+  TransformStream,
+  TransformStreamDefaultController,
+} = require('stream/web');
+
+const {
+  createReadStream,
+  readFileSync,
+} = require('fs');
+
+const {
+  kTransfer,
+} = require('internal/worker/js_transferable');
+
+const {
+  inspect,
+} = require('util');
+
+assert.throws(() => new TransformStream({ readableType: 1 }), {
+  code: 'ERR_INVALID_ARG_VALUE',
+});
+assert.throws(() => new TransformStream({ writableType: 1 }), {
+  code: 'ERR_INVALID_ARG_VALUE',
+});
+
+
+{
+  const stream = new TransformStream();
+
+  async function test(stream) {
+    const writer = stream.writable.getWriter();
+    const reader = stream.readable.getReader();
+
+    const { 1: result } = await Promise.all([
+      writer.write('hello'),
+      reader.read(),
+    ]);
+
+    assert.strictEqual(result.value, 'hello');
+  }
+
+  test(stream).then(common.mustCall());
+}
+
+class Transform {
+  start(controller) {
+    this.started = true;
+  }
+
+  async transform(chunk, controller) {
+    controller.enqueue(chunk.toUpperCase());
+  }
+
+  async flush() {
+    this.flushed = true;
+  }
+}
+
+{
+  const transform = new Transform();
+  const stream = new TransformStream(transform);
+  assert(transform.started);
+
+  async function test(stream) {
+    const writer = stream.writable.getWriter();
+    const reader = stream.readable.getReader();
+
+    const { 1: result } = await Promise.all([
+      writer.write('hello'),
+      reader.read(),
+    ]);
+
+    assert.strictEqual(result.value, 'HELLO');
+
+    await writer.close();
+  }
+
+  test(stream).then(common.mustCall(() => {
+    assert(transform.flushed);
+  }));
+}
+
+class Source {
+  constructor() {
+    this.cancelCalled = false;
+  }
+
+  start(controller) {
+    this.stream = createReadStream(__filename);
+    this.stream.on('data', (chunk) => {
+      controller.enqueue(chunk.toString());
+    });
+    this.stream.once('end', () => {
+      if (!this.cancelCalled)
+        controller.close();
+    });
+    this.stream.once('error', (error) => {
+      controller.error(error);
+    });
+  }
+
+  cancel() {
+    this.cancelCalled = true;
+  }
+}
+
+{
+  const instream = new ReadableStream(new Source());
+  const tstream = new TransformStream(new Transform());
+  const r = instream.pipeThrough(tstream);
+
+  async function read(stream) {
+    let res = '';
+    for await (const chunk of stream)
+      res += chunk;
+    return res;
+  }
+
+  read(r).then(common.mustCall((data) => {
+    const check = readFileSync(__filename);
+    assert.strictEqual(check.toString().toUpperCase(), data);
+  }));
+}
+
+{
+  assert.throws(() => Reflect.get(TransformStream.prototype, 'readable', {}), {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.throws(() => Reflect.get(TransformStream.prototype, 'writable', {}), {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.throws(() => TransformStream.prototype[kTransfer]({}), {
+    code: 'ERR_INVALID_THIS',
+  });
+
+  assert.throws(() => {
+    Reflect.get(TransformStreamDefaultController.prototype, 'desiredSize', {});
+  }, {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.throws(() => {
+    TransformStreamDefaultController.prototype.enqueue({});
+  }, {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.throws(() => {
+    TransformStreamDefaultController.prototype.error({});
+  }, {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.throws(() => {
+    TransformStreamDefaultController.prototype.terminate({});
+  }, {
+    code: 'ERR_INVALID_THIS',
+  });
+
+  assert.throws(() => new TransformStreamDefaultController(), {
+    code: 'ERR_ILLEGAL_CONSTRUCTOR',
+  });
+}
+
+{
+  let controller;
+  const transform = new TransformStream({
+    start(c) {
+      controller = c;
+    }
+  });
+
+  assert.match(inspect(transform), /TransformStream/);
+  assert.match(inspect(transform, { depth: null }), /TransformStream/);
+  assert.match(inspect(transform, { depth: 0 }), /TransformStream \[/);
+
+  assert.match(inspect(controller), /TransformStreamDefaultController/);
+  assert.match(
+    inspect(controller, { depth: null }),
+    /TransformStreamDefaultController/);
+  assert.match(
+    inspect(controller, { depth: 0 }),
+    /TransformStreamDefaultController \[/);
+}

test/parallel/test-whatwg-webstreams-coverage.js

@@ -0,0 +1,70 @@
+// Flags: --no-warnings --expose-internals
+'use strict';
+
+require('../common');
+
+const {
+  ByteLengthQueuingStrategy,
+  CountQueuingStrategy,
+} = require('stream/web');
+
+const {
+  inspect,
+} = require('util');
+
+const {
+  isPromisePending,
+} = require('internal/webstreams/util');
+
+const assert = require('assert');
+
+assert(!isPromisePending({}));
+assert(!isPromisePending(Promise.resolve()));
+assert(isPromisePending(new Promise(() => {})));
+
+// Brand checking works
+assert.throws(() => {
+  Reflect.get(ByteLengthQueuingStrategy.prototype, 'highWaterMark', {});
+}, {
+  code: 'ERR_INVALID_THIS'
+});
+
+assert.throws(() => {
+  Reflect.get(ByteLengthQueuingStrategy.prototype, 'size', {});
+}, {
+  code: 'ERR_INVALID_THIS'
+});
+
+assert.throws(() => {
+  Reflect.get(CountQueuingStrategy.prototype, 'highWaterMark', {});
+}, {
+  code: 'ERR_INVALID_THIS'
+});
+
+assert.throws(() => {
+  Reflect.get(CountQueuingStrategy.prototype, 'size', {});
+}, {
+  code: 'ERR_INVALID_THIS'
+});
+
+// Custom Inspect Works
+
+{
+  const strategy = new CountQueuingStrategy({ highWaterMark: 1 });
+
+  assert.strictEqual(
+    inspect(strategy, { depth: null }),
+    'CountQueuingStrategy { highWaterMark: 1 }');
+
+  assert.strictEqual(
+    inspect(strategy),
+    'CountQueuingStrategy { highWaterMark: 1 }');
+
+  assert.strictEqual(
+    inspect(strategy, { depth: 0 }),
+    'CountQueuingStrategy [Object]');
+
+  assert.strictEqual(
+    inspect(new ByteLengthQueuingStrategy({ highWaterMark: 1 })),
+    'ByteLengthQueuingStrategy { highWaterMark: 1 }');
+}

test/parallel/test-whatwg-webstreams-transfer.js

@@ -0,0 +1,503 @@
+// Flags: --expose-internals --no-warnings
+'use strict';
+
+const common = require('../common');
+
+const {
+  ReadableStream,
+  WritableStream,
+  TransformStream,
+} = require('stream/web');
+
+const {
+  Worker
+} = require('worker_threads');
+
+const {
+  isReadableStream,
+} = require('internal/webstreams/readablestream');
+
+const {
+  isWritableStream,
+} = require('internal/webstreams/writablestream');
+
+const {
+  isTransformStream,
+} = require('internal/webstreams/transformstream');
+
+const {
+  makeTransferable,
+  kClone,
+  kTransfer,
+  kDeserialize,
+} = require('internal/worker/js_transferable');
+
+const assert = require('assert');
+
+const theData = 'hello';
+
+{
+  const { port1, port2 } = new MessageChannel();
+  port1.onmessageerror = common.mustNotCall();
+  port2.onmessageerror = common.mustNotCall();
+
+  // This test takes the ReadableStream and transfers it to the
+  // port1 first, then again to port2, which reads the data.
+  // Internally, this sets up a pipelined data flow that is
+  // important to understand in case this test fails..
+  //
+  // Specifically:
+  //
+  // 1. We start with ReadableStream R1,
+  // 2. Calling port2.postMessage causes a new internal WritableStream W1
+  //    and a new ReadableStream R2 to be created, both of which are coupled
+  //    to each other via a pair of MessagePorts P1 and P2.
+  // 3. ReadableStream R2 is passed to the port1.onmessage callback as the
+  //    data property of the MessageEvent, and R1 is configured to pipeTo W1.
+  // 4. Within port1.onmessage, we transfer ReadableStream R2 to port1, which
+  //    creates a new internal WritableStream W2 and a new ReadableStream R3,
+  //    both of which are coupled to each other via a pair of MessagePorts
+  //    P3 and P4.
+  // 5. ReadableStream R3 is passed to the port2.onmessage callback as the
+  //    data property of the MessageEvent, and R2 is configured to pipeTo W2.
+  // 6. Once the reader is attached to R3 in the port2.onmessage callback,
+  //    a message is sent along the path: R3 -> P4 -> P3 -> R2 -> P2 -> P1 -> R1
+  //    to begin pulling the data. The data is then pushed along the pipeline
+  //    R1 -> W1 -> P1 -> P2 -> R2 -> W2 -> P3 -> P4 -> R3
+  // 7. The MessagePorts P1, P2, P3, and P4 serve as a control channel for
+  //    passing data and control instructions, potentially across realms,
+  //    to the other ReadableStream and WritableStream instances.
+  //
+  // If this test experiences timeouts (hangs without finishing), it's most
+  // likely because the control instructions are somehow broken and the
+  // MessagePorts are not being closed properly or it could be caused by
+  // failing the close R1's controller which signals the end of the data
+  // flow.
+
+  const readable = new ReadableStream({
+    start: common.mustCall((controller) => {
+      controller.enqueue(theData);
+      controller.close();
+    }),
+  });
+
+  port2.onmessage = common.mustCall(({ data }) => {
+    assert(isReadableStream(data));
+
+    const reader = data.getReader();
+    reader.read().then(common.mustCall((chunk) => {
+      assert.deepStrictEqual(chunk, { done: false, value: theData });
+    }));
+
+    port2.close();
+  });
+
+  port1.onmessage = common.mustCall(({ data }) => {
+    assert(isReadableStream(data));
+    assert(!data.locked);
+    port1.postMessage(data, [data]);
+    assert(data.locked);
+  });
+
+  assert.throws(() => port2.postMessage(readable), {
+    code: 'ERR_MISSING_TRANSFERABLE_IN_TRANSFER_LIST',
+  });
+
+  port2.postMessage(readable, [readable]);
+  assert(readable.locked);
+}
+
+{
+  const { port1, port2 } = new MessageChannel();
+  port1.onmessageerror = common.mustNotCall();
+  port2.onmessageerror = common.mustNotCall();
+
+  // Like the ReadableStream test above, this sets up a pipeline
+  // through which the data flows...
+  //
+  // We start with WritableStream W1, which is transfered to port1.
+  // Doing so creates an internal ReadableStream R1 and WritableStream W2,
+  // which are coupled together with MessagePorts P1 and P2.
+  // The port1.onmessage callback receives WritableStream W2 and
+  // immediately transfers that to port2. Doing so creates an internal
+  // ReadableStream R2 and WritableStream W3, which are coupled together
+  // with MessagePorts P3 and P4. WritableStream W3 is handed off to
+  // port2.onmessage.
+  //
+  // When the writer on port2.onmessage writes the chunk of data, it
+  // gets passed along the pipeline:
+  // W3 -> P4 -> P3 -> R2 -> W2 -> P2 -> P1 -> R1 -> W1
+
+  const writable = new WritableStream({
+    write: common.mustCall((chunk) => {
+      assert.strictEqual(chunk, theData);
+    }),
+  });
+
+  port2.onmessage = common.mustCall(({ data }) => {
+    assert(isWritableStream(data));
+    assert(!data.locked);
+    const writer = data.getWriter();
+    writer.write(theData).then(common.mustCall());
+    writer.close();
+    port2.close();
+  });
+
+  port1.onmessage = common.mustCall(({ data }) => {
+    assert(isWritableStream(data));
+    assert(!data.locked);
+    port1.postMessage(data, [data]);
+    assert(data.locked);
+  });
+
+  assert.throws(() => port2.postMessage(writable), {
+    code: 'ERR_MISSING_TRANSFERABLE_IN_TRANSFER_LIST',
+  });
+
+  port2.postMessage(writable, [writable]);
+  assert(writable.locked);
+}
+
+{
+  const { port1, port2 } = new MessageChannel();
+  port1.onmessageerror = common.mustNotCall();
+  port2.onmessageerror = common.mustNotCall();
+
+  // The data flow here is actually quite complicated, and is a combination
+  // of the WritableStream and ReadableStream examples above.
+  //
+  // We start with TransformStream T1, which creates ReadableStream R1,
+  // and WritableStream W1.
+  //
+  // When T1 is transfered to port1.onmessage, R1 and W1 are individually
+  // transfered.
+  //
+  // When R1 is transfered, it creates internal WritableStream W2, and
+  // new ReadableStream R2, coupled together via MessagePorts P1 and P2.
+  //
+  // When W1 is transfered, it creates internal ReadableStream R3 and
+  // new WritableStream W3, coupled together via MessagePorts P3 and P4.
+  //
+  // A new TransformStream T2 is created that owns ReadableStream R2 and
+  // WritableStream W3. The port1.onmessage callback immediately transfers
+  // that to port2.onmessage.
+  //
+  // When T2 is transfered, R2 and W3 are individually transfered.
+  //
+  // When R2 is transfered, it creates internal WritableStream W4, and
+  // ReadableStream R4, coupled together via MessagePorts P5 and P6.
+  //
+  // When W3 is transfered, it creates internal ReadableStream R5, and
+  // WritableStream W5, coupled together via MessagePorts P7 and P8.
+  //
+  // A new TransformStream T3 is created that owns ReadableStream R4 and
+  // WritableStream W5.
+  //
+  // port1.onmessage then writes a chunk of data. That chunk of data
+  // flows through the pipeline to T1:
+  //
+  // W5 -> P8 -> P7 -> R5 -> W3 -> P4 -> P3 -> R3 -> W1 -> T1
+  //
+  // T1 performs the transformation, then pushes the chunk back out
+  // along the pipeline:
+  //
+  // T1 -> R1 -> W2 -> P1 -> P2 -> R2 -> W4 -> P5 -> P6 -> R4
+
+  const transform = new TransformStream({
+    transform(chunk, controller) {
+      controller.enqueue(chunk.toUpperCase());
+    }
+  });
+
+  port2.onmessage = common.mustCall(({ data }) => {
+    assert(isTransformStream(data));
+    const writer = data.writable.getWriter();
+    const reader = data.readable.getReader();
+    Promise.all([
+      writer.write(theData),
+      writer.close(),
+      reader.read().then(common.mustCall((result) => {
+        assert(!result.done);
+        assert.strictEqual(result.value, theData.toUpperCase());
+      })),
+      reader.read().then(common.mustCall((result) => {
+        assert(result.done);
+      })),
+    ]).then(common.mustCall());
+    port2.close();
+  });
+
+  port1.onmessage = common.mustCall(({ data }) => {
+    assert(isTransformStream(data));
+    assert(!data.readable.locked);
+    assert(!data.writable.locked);
+    port1.postMessage(data, [data]);
+    assert(data.readable.locked);
+    assert(data.writable.locked);
+  });
+
+  assert.throws(() => port2.postMessage(transform), {
+    code: 'ERR_MISSING_TRANSFERABLE_IN_TRANSFER_LIST',
+  });
+
+  port2.postMessage(transform, [transform]);
+  assert(transform.readable.locked);
+  assert(transform.writable.locked);
+}
+
+{
+  const { port1, port2 } = new MessageChannel();
+  let controller;
+
+  const readable = new ReadableStream({
+    start(c) { controller = c; },
+
+    cancel: common.mustCall((error) => {
+      assert.strictEqual(error.code, 25);  // DataCloneError
+    }),
+  });
+
+  port1.onmessage = ({ data }) => {
+    const reader = data.getReader();
+    assert.rejects(reader.read(), {
+      code: 25,  // DataCloneError
+    });
+    port1.close();
+  };
+
+  port2.postMessage(readable, [readable]);
+
+  const notActuallyTransferable = makeTransferable({
+    [kClone]() {
+      return {
+        data: {},
+        deserializeInfo: 'nothing that will work',
+      };
+    },
+    [kDeserialize]: common.mustNotCall(),
+  });
+
+  controller.enqueue(notActuallyTransferable);
+}
+
+{
+  const { port1, port2 } = new MessageChannel();
+
+  const source = {
+    abort: common.mustCall((error) => {
+      process.nextTick(() => {
+        assert.strictEqual(error.code, 25);
+        assert.strictEqual(error.name, 'DataCloneError');
+      });
+    })
+  };
+
+  const writable = new WritableStream(source);
+
+  const notActuallyTransferable = makeTransferable({
+    [kClone]() {
+      return {
+        data: {},
+        deserializeInfo: 'nothing that will work',
+      };
+    },
+    [kDeserialize]: common.mustNotCall(),
+  });
+
+  port1.onmessage = common.mustCall(({ data }) => {
+    const writer = data.getWriter();
+
+    assert.rejects(writer.closed, {
+      code: 25,
+      name: 'DataCloneError',
+    });
+
+    writer.write(notActuallyTransferable).then(common.mustCall());
+
+    port1.close();
+  });
+
+  port2.postMessage(writable, [writable]);
+}
+
+{
+  const error = new Error('boom');
+  const { port1, port2 } = new MessageChannel();
+
+  const source = {
+    abort: common.mustCall((reason) => {
+      process.nextTick(() => {
+        assert.deepStrictEqual(reason, error);
+
+        // Reason is a clone of the original error.
+        assert.notStrictEqual(reason, error);
+      });
+    }),
+  };
+
+  const writable = new WritableStream(source);
+
+  port1.onmessage = common.mustCall(({ data }) => {
+    const writer = data.getWriter();
+
+    assert.rejects(writer.closed, error);
+
+    writer.abort(error).then(common.mustCall());
+    port1.close();
+  });
+
+  port2.postMessage(writable, [writable]);
+}
+
+{
+  const { port1, port2 } = new MessageChannel();
+
+  const source = {
+    abort: common.mustCall((error) => {
+      process.nextTick(() => assert.strictEqual(error.code, 25));
+    })
+  };
+
+  const writable = new WritableStream(source);
+
+  port1.onmessage = common.mustCall(({ data }) => {
+    const writer = data.getWriter();
+
+    const m = new WebAssembly.Memory({ initial: 1 });
+
+    assert.rejects(writer.abort(m), {
+      code: 25
+    });
+    port1.close();
+  });
+
+  port2.postMessage(writable, [writable]);
+}
+
+{
+  // Verify that the communication works across worker threads...
+
+  const worker = new Worker(`
+    const {
+      isReadableStream,
+    } = require('internal/webstreams/readablestream');
+
+    const {
+      parentPort,
+    } = require('worker_threads');
+
+    const assert = require('assert');
+
+    const tracker = new assert.CallTracker();
+    process.on('exit', () => {
+      tracker.verify();
+    });
+
+    parentPort.onmessage = tracker.calls(({ data }) => {
+      assert(isReadableStream(data));
+      const reader = data.getReader();
+      reader.read().then(tracker.calls((result) => {
+        assert(!result.done);
+        assert(result.value instanceof Uint8Array);
+      }));
+      parentPort.close();
+    });
+    parentPort.onmessageerror = () => assert.fail('should not be called');
+  `, { eval: true });
+
+  worker.on('error', common.mustNotCall());
+
+  const readable = new ReadableStream({
+    start(controller) {
+      controller.enqueue(new Uint8Array(10));
+      controller.close();
+    }
+  });
+
+  worker.postMessage(readable, [readable]);
+}
+
+{
+  const source = {
+    cancel: common.mustCall(),
+  };
+
+  const readable = new ReadableStream(source);
+
+  const { port1, port2 } = new MessageChannel();
+
+  port1.onmessage = common.mustCall(({ data }) => {
+    data.cancel().then(common.mustCall());
+    port1.close();
+  });
+
+  port2.postMessage(readable, [readable]);
+}
+
+{
+  const source = {
+    cancel: common.mustCall((error) => {
+      process.nextTick(() => assert(error.code, 25));
+    }),
+  };
+
+  const readable = new ReadableStream(source);
+
+  const { port1, port2 } = new MessageChannel();
+
+  port1.onmessage = common.mustCall(({ data }) => {
+    const m = new WebAssembly.Memory({ initial: 1 });
+
+    const reader = data.getReader();
+
+    const cancel = reader.cancel(m);
+
+    reader.closed.then(common.mustCall());
+
+    assert.rejects(cancel, {
+      code: 25
+    });
+
+    port1.close();
+  });
+
+  port2.postMessage(readable, [readable]);
+}
+
+{
+  const source = {
+    abort: common.mustCall((error) => {
+      process.nextTick(() => {
+        assert.strictEqual(error.code, 25);
+      });
+    }),
+  };
+
+  const writable = new WritableStream(source);
+
+  const { port1, port2 } = new MessageChannel();
+
+  port1.onmessage = common.mustCall(({ data }) => {
+    const m = new WebAssembly.Memory({ initial: 1 });
+    const writer = data.getWriter();
+    const write = writer.write(m);
+    assert.rejects(write, { code: 25 });
+    port1.close();
+  });
+
+  port2.postMessage(writable, [writable]);
+}
+
+{
+  const readable = new ReadableStream();
+  readable.getReader();
+  assert.throws(() => readable[kTransfer](), {
+    code: 25
+  });
+
+  const writable = new WritableStream();
+  writable.getWriter();
+  assert.throws(() => writable[kTransfer](), {
+    code: 25
+  });
+}

test/parallel/test-whatwg-writablestream.js

@@ -0,0 +1,260 @@
+// Flags: --expose-internals --no-warnings
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+
+const {
+  WritableStream,
+  WritableStreamDefaultController,
+  WritableStreamDefaultWriter,
+  CountQueuingStrategy,
+} = require('stream/web');
+
+const {
+  kState,
+} = require('internal/webstreams/util');
+
+const {
+  isPromise,
+} = require('util/types');
+
+const {
+  kTransfer,
+} = require('internal/worker/js_transferable');
+
+const {
+  inspect,
+} = require('util');
+
+class Sink {
+  constructor() {
+    this.chunks = [];
+  }
+
+  start() {
+    this.started = true;
+  }
+
+  write(chunk) {
+    this.chunks.push(chunk);
+  }
+
+  close() {
+    this.closed = true;
+  }
+
+  abort() {
+    this.aborted = true;
+  }
+}
+
+{
+  const stream = new WritableStream();
+
+  assert(stream[kState].controller instanceof WritableStreamDefaultController);
+  assert(!stream.locked);
+
+  assert.strictEqual(typeof stream.abort, 'function');
+  assert.strictEqual(typeof stream.close, 'function');
+  assert.strictEqual(typeof stream.getWriter, 'function');
+}
+
+[1, false, ''].forEach((type) => {
+  assert.throws(() => new WritableStream({ type }), {
+    code: 'ERR_INVALID_ARG_VALUE',
+  });
+});
+
+['a', {}].forEach((highWaterMark) => {
+  assert.throws(() => new WritableStream({}, { highWaterMark }), {
+    code: 'ERR_INVALID_ARG_VALUE',
+  });
+});
+
+['a', false, {}].forEach((size) => {
+  assert.throws(() => new WritableStream({}, { size }), {
+    code: 'ERR_INVALID_ARG_TYPE',
+  });
+});
+
+{
+  new WritableStream({}, 1);
+  new WritableStream({}, 'a');
+  new WritableStream({}, null);
+}
+
+{
+  const sink = new Sink();
+  const stream = new WritableStream(
+    sink,
+    new CountQueuingStrategy({ highWaterMark: 1 }));
+
+  assert(!stream.locked);
+  const writer = stream.getWriter();
+  assert(stream.locked);
+  assert(writer instanceof WritableStreamDefaultWriter);
+
+  assert(isPromise(writer.closed));
+  assert(isPromise(writer.ready));
+  assert(typeof writer.desiredSize, 'number');
+  assert(typeof writer.abort, 'function');
+  assert(typeof writer.close, 'function');
+  assert(typeof writer.releaseLock, 'function');
+  assert(typeof writer.write, 'function');
+
+  writer.releaseLock();
+  assert(!stream.locked);
+
+  const writer2 = stream.getWriter();
+
+  assert(sink.started);
+
+  writer2.closed.then(common.mustCall());
+  writer2.ready.then(common.mustCall());
+
+  writer2.close().then(common.mustCall(() => {
+    assert.strict(sink.closed);
+  }));
+}
+
+{
+  const sink = new Sink();
+
+  const stream = new WritableStream(
+    sink,
+    new CountQueuingStrategy({ highWaterMark: 1 }));
+
+  const error = new Error('boom');
+
+  const writer = stream.getWriter();
+
+  assert.rejects(writer.closed, error);
+
+  writer.abort(error).then(common.mustCall(() => {
+    assert.strictEqual(stream[kState].state, 'errored');
+    assert(sink.aborted);
+  }));
+}
+
+{
+  const sink = new Sink();
+
+  const stream = new WritableStream(
+    sink, { highWaterMark: 1 }
+  );
+
+  async function write(stream) {
+    const writer = stream.getWriter();
+    const p = writer.write('hello');
+    assert.strictEqual(writer.desiredSize, 0);
+    await p;
+    assert.strictEqual(writer.desiredSize, 1);
+  }
+
+  write(stream).then(common.mustCall(() => {
+    assert.deepStrictEqual(['hello'], sink.chunks);
+  }));
+}
+
+{
+  assert.throws(() => Reflect.get(WritableStream.prototype, 'locked', {}), {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.rejects(() => WritableStream.prototype.abort({}), {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.rejects(() => WritableStream.prototype.close({}), {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.throws(() => WritableStream.prototype.getWriter.call(), {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.throws(() => WritableStream.prototype[kTransfer].call(), {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.rejects(
+    Reflect.get(WritableStreamDefaultWriter.prototype, 'closed'), {
+      code: 'ERR_INVALID_THIS',
+    });
+  assert.rejects(
+    Reflect.get(WritableStreamDefaultWriter.prototype, 'ready'), {
+      code: 'ERR_INVALID_THIS',
+    });
+  assert.throws(
+    () => Reflect.get(WritableStreamDefaultWriter.prototype, 'desiredSize'), {
+      code: 'ERR_INVALID_THIS',
+    });
+  assert.rejects(WritableStreamDefaultWriter.prototype.abort({}), {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.rejects(WritableStreamDefaultWriter.prototype.close({}), {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.rejects(WritableStreamDefaultWriter.prototype.write({}), {
+    code: 'ERR_INVALID_THIS',
+  });
+  assert.throws(() => WritableStreamDefaultWriter.prototype.releaseLock({}), {
+    code: 'ERR_INVALID_THIS',
+  });
+
+  assert.throws(() => {
+    Reflect.get(WritableStreamDefaultController.prototype, 'abortReason', {});
+  }, {
+    code: 'ERR_INVALID_THIS',
+  });
+
+  assert.throws(() => {
+    Reflect.get(WritableStreamDefaultController.prototype, 'signal', {});
+  }, {
+    code: 'ERR_INVALID_THIS',
+  });
+
+  assert.throws(() => {
+    WritableStreamDefaultController.prototype.error({});
+  }, {
+    code: 'ERR_INVALID_THIS',
+  });
+}
+
+{
+  let controller;
+  const writable = new WritableStream({
+    start(c) { controller = c; }
+  });
+  assert.strictEqual(
+    inspect(writable),
+    'WritableStream { locked: false, state: \'writable\' }');
+  assert.strictEqual(
+    inspect(writable, { depth: null }),
+    'WritableStream { locked: false, state: \'writable\' }');
+  assert.strictEqual(
+    inspect(writable, { depth: 0 }),
+    'WritableStream [Object]');
+
+  const writer = writable.getWriter();
+  assert.match(
+    inspect(writer),
+    /WritableStreamDefaultWriter/);
+  assert.match(
+    inspect(writer, { depth: null }),
+    /WritableStreamDefaultWriter/);
+  assert.match(
+    inspect(writer, { depth: 0 }),
+    /WritableStreamDefaultWriter \[/);
+
+  assert.match(
+    inspect(controller),
+    /WritableStreamDefaultController/);
+  assert.match(
+    inspect(controller, { depth: null }),
+    /WritableStreamDefaultController/);
+  assert.match(
+    inspect(controller, { depth: 0 }),
+    /WritableStreamDefaultController \[/);
+
+  writer.abort(new Error('boom'));
+
+  assert.strictEqual(writer.desiredSize, null);
+  setImmediate(() => assert.strictEqual(writer.desiredSize, null));
+}

test/wpt/status/streams.json

@@ -1 +1,11 @@
-{}
+{
+  "queuing-strategies-size-function-per-global.window.js": {
+    "skip": "Browser-specific test"
+  },
+  "transferable/deserialize-error.window.js": {
+    "skip": "Browser-specific test"
+  },
+  "readable-byte-streams/bad-buffers-and-views.any.js": {
+    "fail": "TODO: implement detached ArrayBuffer support"
+  }
+}

test/wpt/test-streams.js

@@ -10,7 +10,7 @@ runner.setFlags(['--expose-internals']);
 
 // Set a script that will be executed in the worker before running the tests.
 runner.setInitScript(`
-  const {
+  let {
     ReadableStream,
     ReadableStreamDefaultReader,
     ReadableStreamBYOBReader,
@@ -29,19 +29,111 @@ runner.setInitScript(`
   const { internalBinding } = require('internal/test/binding');
   const { DOMException } = internalBinding('messaging');
   global.DOMException = DOMException;
-  global.ReadableStream = ReadableStream;
-  global.ReadableStreamDefaultReader = ReadableStreamDefaultReader;
-  global.ReadableStreamBYOBReader = ReadableStreamBYOBReader;
-  global.ReadableStreamBYOBRequest = ReadableStreamBYOBRequest;
-  global.ReadableByteStreamController = ReadableByteStreamController;
-  global.ReadableStreamDefaultController = ReadableStreamDefaultController;
-  global.TransformStream = TransformStream;
-  global.TransformStreamDefaultController = TransformStreamDefaultController;
-  global.WritableStream = WritableStream;
-  global.WritableStreamDefaultWriter = WritableStreamDefaultWriter;
-  global.WritableStreamDefaultController = WritableStreamDefaultController;
-  global.ByteLengthQueuingStrategy = ByteLengthQueuingStrategy;
-  global.CountQueuingStrategy = CountQueuingStrategy;
+
+  Object.defineProperties(global, {
+    ReadableStream: {
+      value: ReadableStream,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    ReadableStreamDefaultReader: {
+      value: ReadableStreamDefaultReader,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    ReadableStreamBYOBReader: {
+      value: ReadableStreamBYOBReader,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    ReadableStreamBYOBRequest: {
+      value: ReadableStreamBYOBRequest,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    ReadableByteStreamController: {
+      value: ReadableByteStreamController,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    ReadableStreamDefaultController: {
+      value: ReadableStreamDefaultController,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    TransformStream: {
+      value: TransformStream,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    TransformStreamDefaultController: {
+      value: TransformStreamDefaultController,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    WritableStream: {
+      value: WritableStream,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    WritableStreamDefaultWriter: {
+      value: WritableStreamDefaultWriter,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    WritableStreamDefaultController: {
+      value: WritableStreamDefaultController,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    ByteLengthQueuingStrategy: {
+      value: ByteLengthQueuingStrategy,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+    CountQueuingStrategy: {
+      value: CountQueuingStrategy,
+      configurable: true,
+      writable: true,
+      enumerable: false,
+    },
+  });
+
+  // Simulate global postMessage for enqueue-with-detached-buffer.window.js
+  function postMessage(value, origin, transferList) {
+    const mc = new MessageChannel();
+    mc.port1.postMessage(value, transferList);
+    mc.port2.close();
+  }
+
+  // TODO(@jasnell): This is a bit of a hack to get the idl harness test
+  // working. Later we should investigate a better approach.
+  // See: https://github.com/nodejs/node/pull/39062#discussion_r659383373
+  Object.defineProperties(global, {
+    DedicatedWorkerGlobalScope: {
+      get() {
+        // Pretend that we're a DedicatedWorker, but *only* for the
+        // IDL harness. For everything else, keep the JavaScript shell
+        // environment.
+        if (new Error().stack.includes('idlharness.js'))
+          return global.constructor;
+        else
+          return function() {};
+      }
+    }
+  });
 `);
 
 runner.runJsTests();

tools/doc/type-parser.mjs

@@ -226,6 +226,33 @@ const customTypesMap = {
   'X509Certificate': 'crypto.html#crypto_class_x509certificate',
 
   'zlib options': 'zlib.html#zlib_class_options',
+
+  'ReadableStream':
+    'webstreams.md#webstreamsapi_class_readablestream',
+  'ReadableStreamDefaultReader':
+    'webstreams.md#webstreamsapi_class_readablestreamdefaultreader',
+  'ReadableStreamBYOBReader':
+    'webstreams.md#webstreamsapi_class_readablestreambyobreader',
+  'ReadableStreamDefaultController':
+    'webstreams.md#webstreamsapi_class_readablestreamdefaultcontroller',
+  'ReadableByteStreamController':
+    'webstreams.md#webstreamsapi_class_readablebytestreamcontroller',
+  'ReadableStreamBYOBRequest':
+    'webstreams.md#webstreamsapi_class_readablestreambyobrequest',
+  'WritableStream':
+    'webstreams.md#webstreamsapi_class_writablestream',
+  'WritableStreamDefaultWriter':
+    'webstreams.md#webstreamsapi_class_writablestreamdefaultwriter',
+  'WritableStreamDefaultController':
+    'webstreams.md#webstreamsapi_class_writablestreamdefaultcontroller',
+  'TransformStream':
+    'webstreams.md#webstreamsapi_class_transformstream',
+  'TransformStreamDefaultController':
+    'webstreams.md#webstreamsapi_class_transformstreamdefaultcontroller',
+  'ByteLengthQueuingStrategy':
+    'webstreams.md#webstreamsapi_class_bytelengthqueuingstrategy',
+  'CountQueuingStrategy':
+    'webstreams.md#webstreamsapi_class_countqueuingstrategy',
 };
 
 const arrayPart = /(?:\[])+$/;

typings/primordials.d.ts

@@ -1,3 +1,5 @@
+import { AsyncIterator } from "internal/webstreams/util";
+
 type UncurryThis<T extends (this: unknown, ...args: unknown[]) => unknown> =
   (self: ThisParameterType<T>, ...args: Parameters<T>) => ReturnType<T>;
 type UncurryThisStaticApply<T extends (this: unknown, ...args: unknown[]) => unknown> =
@@ -9,15 +11,15 @@ type StaticApply<T extends (this: unknown, ...args: unknown[]) => unknown> =
  * Primordials are a way to safely use globals without fear of global mutation
  * Generally, this means removing `this` parameter usage and instead using
  * a regular parameter:
- * 
+ *
  * @example
- * 
+ *
  * ```js
  * 'thing'.startsWith('hello');
  * ```
- * 
+ *
  * becomes
- * 
+ *
  * ```js
  * primordials.StringPrototypeStartsWith('thing', 'hello')
  * ```
@@ -142,6 +144,7 @@ declare namespace primordials {
   export const ArrayBufferPrototype: typeof ArrayBuffer.prototype
   export const ArrayBufferIsView: typeof ArrayBuffer.isView
   export const ArrayBufferPrototypeSlice: UncurryThis<typeof ArrayBuffer.prototype.slice>
+  export const AsyncIteratorPrototype: UncurryThis<typeof AsyncIterator>
   export const BigInt: typeof globalThis.BigInt;
   export const BigIntLength: typeof BigInt.length
   export const BigIntName: typeof BigInt.name
@@ -522,5 +525,5 @@ declare namespace primordials {
   export const PromiseAny: typeof Promise.any
   export const PromisePrototypeThen: UncurryThis<typeof Promise.prototype.then>
   export const PromisePrototypeCatch: UncurryThis<typeof Promise.prototype.catch>
-  export const PromisePrototypeFinally: UncurryThis<typeof Promise.prototype.finally>  
+  export const PromisePrototypeFinally: UncurryThis<typeof Promise.prototype.finally>
 }