From b361ad72ce65894a250fd0543b0e09d1a9a7edc5 Mon Sep 17 00:00:00 2001
From: Debadree Chatterjee <debadree333@gmail.com>
Date: Tue, 18 Jul 2023 15:20:22 +0530
Subject: [PATCH] stream: implement ReadableStream.from

Fixes: https://github.com/nodejs/node/issues/48389
PR-URL: https://github.com/nodejs/node/pull/48395
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Matthew Aitken <maitken033380023@gmail.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
---
 doc/api/webstreams.md                         |  43 ++
 lib/internal/webstreams/readablestream.js     |  59 +++
 lib/internal/webstreams/util.js               |  53 ++
 test/fixtures/wpt/README.md                   |   2 +-
 .../streams/piping/general-addition.any.js    |  15 +
 .../wpt/streams/readable-streams/from.any.js  | 474 ++++++++++++++++++
 .../owning-type-message-port.any.js           |  49 ++
 .../owning-type-video-frame.any.js            | 128 +++++
 .../readable-streams/owning-type.any.js       |  91 ++++
 .../read-task-handling.window.js              |  46 ++
 .../transform-stream-members.any.js           |  16 +
 .../invalid-realm.tentative.window.js         |  17 +
 test/fixtures/wpt/versions.json               |   2 +-
 test/wpt/status/streams.json                  |  43 ++
 14 files changed, 1036 insertions(+), 2 deletions(-)
 create mode 100644 test/fixtures/wpt/streams/piping/general-addition.any.js
 create mode 100644 test/fixtures/wpt/streams/readable-streams/from.any.js
 create mode 100644 test/fixtures/wpt/streams/readable-streams/owning-type-message-port.any.js
 create mode 100644 test/fixtures/wpt/streams/readable-streams/owning-type-video-frame.any.js
 create mode 100644 test/fixtures/wpt/streams/readable-streams/owning-type.any.js
 create mode 100644 test/fixtures/wpt/streams/readable-streams/read-task-handling.window.js
 create mode 100644 test/fixtures/wpt/streams/transferable/transform-stream-members.any.js
 create mode 100644 test/fixtures/wpt/streams/transform-streams/invalid-realm.tentative.window.js

diff --git a/doc/api/webstreams.md b/doc/api/webstreams.md
index a413beb7174fd4..bb3374e1480e73 100644
--- a/doc/api/webstreams.md
+++ b/doc/api/webstreams.md
@@ -387,6 +387,49 @@ port1.onmessage = ({ data }) => {
 port2.postMessage(stream, [stream]);
 ```
 
+### `ReadableStream.from(iterable)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `iterable` {Iterable} Object implementing the `Symbol.asyncIterator` or
+  `Symbol.iterator` iterable protocol.
+
+A utility method that creates a new {ReadableStream} from an iterable.
+
+```mjs
+import { ReadableStream } from 'node:stream/web';
+
+async function* asyncIterableGenerator() {
+  yield 'a';
+  yield 'b';
+  yield 'c';
+}
+
+const stream = ReadableStream.from(asyncIterableGenerator());
+
+for await (const chunk of stream)
+  console.log(chunk); // Prints 'a', 'b', 'c'
+```
+
+```cjs
+const { ReadableStream } = require('node:stream/web');
+
+async function* asyncIterableGenerator() {
+  yield 'a';
+  yield 'b';
+  yield 'c';
+}
+
+(async () => {
+  const stream = ReadableStream.from(asyncIterableGenerator());
+
+  for await (const chunk of stream)
+    console.log(chunk); // Prints 'a', 'b', 'c'
+})();
+```
+
 ### Class: `ReadableStreamDefaultReader`
 
 <!-- YAML
diff --git a/lib/internal/webstreams/readablestream.js b/lib/internal/webstreams/readablestream.js
index 9e754309557b49..e7d115e98f12b9 100644
--- a/lib/internal/webstreams/readablestream.js
+++ b/lib/internal/webstreams/readablestream.js
@@ -111,6 +111,8 @@ const {
   nonOpCancel,
   nonOpPull,
   nonOpStart,
+  getIterator,
+  iteratorNext,
   kType,
   kState,
 } = require('internal/webstreams/util');
@@ -314,6 +316,10 @@ class ReadableStream {
     return isReadableStreamLocked(this);
   }
 
+  static from(iterable) {
+    return readableStreamFromIterable(iterable);
+  }
+
   /**
    * @param {any} [reason]
    * @returns { Promise<void> }
@@ -1250,6 +1256,59 @@ const isReadableStreamBYOBReader =
 
 // ---- ReadableStream Implementation
 
+function readableStreamFromIterable(iterable) {
+  let stream;
+  const iteratorRecord = getIterator(iterable, 'async');
+
+  const startAlgorithm = nonOpStart;
+
+  async function pullAlgorithm() {
+    const nextResult = iteratorNext(iteratorRecord);
+    const nextPromise = PromiseResolve(nextResult);
+    return PromisePrototypeThen(nextPromise, (iterResult) => {
+      if (typeof iterResult !== 'object' || iterResult === null) {
+        throw new ERR_INVALID_STATE.TypeError(
+          'The promise returned by the iterator.next() method must fulfill with an object');
+      }
+      if (iterResult.done) {
+        readableStreamDefaultControllerClose(stream[kState].controller);
+      } else {
+        readableStreamDefaultControllerEnqueue(stream[kState].controller, iterResult.value);
+      }
+    });
+  }
+
+  async function cancelAlgorithm(reason) {
+    const iterator = iteratorRecord.iterator;
+    const returnMethod = iterator.return;
+    if (returnMethod === undefined) {
+      return PromiseResolve();
+    }
+    const returnResult = FunctionPrototypeCall(returnMethod, iterator, reason);
+    const returnPromise = PromiseResolve(returnResult);
+    return PromisePrototypeThen(returnPromise, (iterResult) => {
+      if (typeof iterResult !== 'object' || iterResult === null) {
+        throw new ERR_INVALID_STATE.TypeError(
+          'The promise returned by the iterator.return() method must fulfill with an object');
+      }
+      return undefined;
+    });
+  }
+
+  stream = new ReadableStream({
+    start: startAlgorithm,
+    pull: pullAlgorithm,
+    cancel: cancelAlgorithm,
+  }, {
+    size() {
+      return 1;
+    },
+    highWaterMark: 0,
+  });
+
+  return stream;
+}
+
 function readableStreamPipeTo(
   source,
   dest,
diff --git a/lib/internal/webstreams/util.js b/lib/internal/webstreams/util.js
index abc064170926dd..1979c55667b167 100644
--- a/lib/internal/webstreams/util.js
+++ b/lib/internal/webstreams/util.js
@@ -13,6 +13,8 @@ const {
   PromiseReject,
   ReflectGet,
   Symbol,
+  SymbolAsyncIterator,
+  SymbolIterator,
   Uint8Array,
 } = primordials;
 
@@ -20,6 +22,7 @@ const {
   codes: {
     ERR_INVALID_ARG_VALUE,
     ERR_OPERATION_FAILED,
+    ERR_INVALID_STATE,
   },
 } = require('internal/errors');
 
@@ -217,6 +220,54 @@ function lazyTransfer() {
   return transfer;
 }
 
+function createAsyncFromSyncIterator(syncIteratorRecord) {
+  const syncIterable = {
+    [SymbolIterator]: () => syncIteratorRecord.iterator,
+  };
+
+  const asyncIterator = (async function* () {
+    return yield* syncIterable;
+  }());
+
+  const nextMethod = asyncIterator.next;
+  return { iterator: asyncIterator, nextMethod, done: false };
+}
+
+function getIterator(obj, kind = 'sync', method) {
+  if (method === undefined) {
+    if (kind === 'async') {
+      method = obj[SymbolAsyncIterator];
+      if (method === undefined) {
+        const syncMethod = obj[SymbolIterator];
+        const syncIteratorRecord = getIterator(obj, 'sync', syncMethod);
+        return createAsyncFromSyncIterator(syncIteratorRecord);
+      }
+    } else {
+      method = obj[SymbolIterator];
+    }
+  }
+
+  const iterator = FunctionPrototypeCall(method, obj);
+  if (typeof iterator !== 'object' || iterator === null) {
+    throw new ERR_INVALID_STATE.TypeError('The iterator method must return an object');
+  }
+  const nextMethod = iterator.next;
+  return { iterator, nextMethod, done: false };
+}
+
+function iteratorNext(iteratorRecord, value) {
+  let result;
+  if (value === undefined) {
+    result = FunctionPrototypeCall(iteratorRecord.nextMethod, iteratorRecord.iterator);
+  } else {
+    result = FunctionPrototypeCall(iteratorRecord.nextMethod, iteratorRecord.iterator, [value]);
+  }
+  if (typeof result !== 'object' || result === null) {
+    throw new ERR_INVALID_STATE.TypeError('The iterator.next() method must return an object');
+  }
+  return result;
+}
+
 module.exports = {
   ArrayBufferViewGetBuffer,
   ArrayBufferViewGetByteLength,
@@ -243,6 +294,8 @@ module.exports = {
   nonOpPull,
   nonOpStart,
   nonOpWrite,
+  getIterator,
+  iteratorNext,
   kType,
   kState,
 };
diff --git a/test/fixtures/wpt/README.md b/test/fixtures/wpt/README.md
index 0efe5f719e793c..890c8f9e1dbe3b 100644
--- a/test/fixtures/wpt/README.md
+++ b/test/fixtures/wpt/README.md
@@ -26,7 +26,7 @@ Last update:
 - performance-timeline: https://github.com/web-platform-tests/wpt/tree/17ebc3aea0/performance-timeline
 - resource-timing: https://github.com/web-platform-tests/wpt/tree/22d38586d0/resource-timing
 - resources: https://github.com/web-platform-tests/wpt/tree/919874f84f/resources
-- streams: https://github.com/web-platform-tests/wpt/tree/51750bc8d7/streams
+- streams: https://github.com/web-platform-tests/wpt/tree/517e945bbf/streams
 - url: https://github.com/web-platform-tests/wpt/tree/84782d9315/url
 - user-timing: https://github.com/web-platform-tests/wpt/tree/5ae85bf826/user-timing
 - wasm/jsapi: https://github.com/web-platform-tests/wpt/tree/cde25e7e3c/wasm/jsapi
diff --git a/test/fixtures/wpt/streams/piping/general-addition.any.js b/test/fixtures/wpt/streams/piping/general-addition.any.js
new file mode 100644
index 00000000000000..2562b7064338ee
--- /dev/null
+++ b/test/fixtures/wpt/streams/piping/general-addition.any.js
@@ -0,0 +1,15 @@
+// META: global=window,worker
+'use strict';
+
+promise_test(async t => {
+  /** @type {ReadableStreamDefaultController} */
+  var con;
+  let synchronous = false;
+  new ReadableStream({ start(c) { con = c }}, { highWaterMark: 0 }).pipeTo(
+    new WritableStream({ write() { synchronous = true; } })
+  )
+  // wait until start algorithm finishes
+  await Promise.resolve();
+  con.enqueue();
+  assert_false(synchronous, 'write algorithm must not run synchronously');
+}, "enqueue() must not synchronously call write algorithm");
diff --git a/test/fixtures/wpt/streams/readable-streams/from.any.js b/test/fixtures/wpt/streams/readable-streams/from.any.js
new file mode 100644
index 00000000000000..04a03545ad5fe3
--- /dev/null
+++ b/test/fixtures/wpt/streams/readable-streams/from.any.js
@@ -0,0 +1,474 @@
+// META: global=window,worker,jsshell
+// META: script=../resources/test-utils.js
+'use strict';
+
+const iterableFactories = [
+  ['an array of values', () => {
+    return ['a', 'b'];
+  }],
+
+  ['an array of promises', () => {
+    return [
+      Promise.resolve('a'),
+      Promise.resolve('b')
+    ];
+  }],
+
+  ['an array iterator', () => {
+    return ['a', 'b'][Symbol.iterator]();
+  }],
+
+  ['a string', () => {
+    // This iterates over the code points of the string.
+    return 'ab';
+  }],
+
+  ['a Set', () => {
+    return new Set(['a', 'b']);
+  }],
+
+  ['a Set iterator', () => {
+    return new Set(['a', 'b'])[Symbol.iterator]();
+  }],
+
+  ['a sync generator', () => {
+    function* syncGenerator() {
+      yield 'a';
+      yield 'b';
+    }
+
+    return syncGenerator();
+  }],
+
+  ['an async generator', () => {
+    async function* asyncGenerator() {
+      yield 'a';
+      yield 'b';
+    }
+
+    return asyncGenerator();
+  }],
+
+  ['a sync iterable of values', () => {
+    const chunks = ['a', 'b'];
+    const it = {
+      next() {
+        return {
+          done: chunks.length === 0,
+          value: chunks.shift()
+        };
+      },
+      [Symbol.iterator]: () => it
+    };
+    return it;
+  }],
+
+  ['a sync iterable of promises', () => {
+    const chunks = ['a', 'b'];
+    const it = {
+      next() {
+        return chunks.length === 0 ? { done: true } : {
+          done: false,
+          value: Promise.resolve(chunks.shift())
+        };
+      },
+      [Symbol.iterator]: () => it
+    };
+    return it;
+  }],
+
+  ['an async iterable', () => {
+    const chunks = ['a', 'b'];
+    const it = {
+      next() {
+        return Promise.resolve({
+          done: chunks.length === 0,
+          value: chunks.shift()
+        })
+      },
+      [Symbol.asyncIterator]: () => it
+    };
+    return it;
+  }],
+
+  ['a ReadableStream', () => {
+    return new ReadableStream({
+      start(c) {
+        c.enqueue('a');
+        c.enqueue('b');
+        c.close();
+      }
+    });
+  }],
+
+  ['a ReadableStream async iterator', () => {
+    return new ReadableStream({
+      start(c) {
+        c.enqueue('a');
+        c.enqueue('b');
+        c.close();
+      }
+    })[Symbol.asyncIterator]();
+  }]
+];
+
+for (const [label, factory] of iterableFactories) {
+  promise_test(async () => {
+
+    const iterable = factory();
+    const rs = ReadableStream.from(iterable);
+    assert_equals(rs.constructor, ReadableStream, 'from() should return a ReadableStream');
+
+    const reader = rs.getReader();
+    assert_object_equals(await reader.read(), { value: 'a', done: false }, 'first read should be correct');
+    assert_object_equals(await reader.read(), { value: 'b', done: false }, 'second read should be correct');
+    assert_object_equals(await reader.read(), { value: undefined, done: true }, 'third read should be done');
+    await reader.closed;
+
+  }, `ReadableStream.from accepts ${label}`);
+}
+
+const badIterables = [
+  ['null', null],
+  ['undefined', undefined],
+  ['0', 0],
+  ['NaN', NaN],
+  ['true', true],
+  ['{}', {}],
+  ['Object.create(null)', Object.create(null)],
+  ['a function', () => 42],
+  ['a symbol', Symbol()],
+  ['an object with a non-callable @@iterator method', {
+    [Symbol.iterator]: 42
+  }],
+  ['an object with a non-callable @@asyncIterator method', {
+    [Symbol.asyncIterator]: 42
+  }],
+];
+
+for (const [label, iterable] of badIterables) {
+  test(() => {
+    assert_throws_js(TypeError, () => ReadableStream.from(iterable), 'from() should throw a TypeError')
+  }, `ReadableStream.from throws on invalid iterables; specifically ${label}`);
+}
+
+test(() => {
+  const theError = new Error('a unique string');
+  const iterable = {
+    [Symbol.iterator]() {
+      throw theError;
+    }
+  };
+
+  assert_throws_exactly(theError, () => ReadableStream.from(iterable), 'from() should re-throw the error');
+}, `ReadableStream.from re-throws errors from calling the @@iterator method`);
+
+test(() => {
+  const theError = new Error('a unique string');
+  const iterable = {
+    [Symbol.asyncIterator]() {
+      throw theError;
+    }
+  };
+
+  assert_throws_exactly(theError, () => ReadableStream.from(iterable), 'from() should re-throw the error');
+}, `ReadableStream.from re-throws errors from calling the @@asyncIterator method`);
+
+test(t => {
+  const theError = new Error('a unique string');
+  const iterable = {
+    [Symbol.iterator]: t.unreached_func('@@iterator should not be called'),
+    [Symbol.asyncIterator]() {
+      throw theError;
+    }
+  };
+
+  assert_throws_exactly(theError, () => ReadableStream.from(iterable), 'from() should re-throw the error');
+}, `ReadableStream.from ignores @@iterator if @@asyncIterator exists`);
+
+promise_test(async () => {
+
+  const iterable = {
+    async next() {
+      return { value: undefined, done: true };
+    },
+    [Symbol.asyncIterator]: () => iterable
+  };
+
+  const rs = ReadableStream.from(iterable);
+  const reader = rs.getReader();
+
+  const read = await reader.read();
+  assert_object_equals(read, { value: undefined, done: true }, 'first read should be done');
+
+  await reader.closed;
+
+}, `ReadableStream.from accepts an empty iterable`);
+
+promise_test(async t => {
+
+  const theError = new Error('a unique string');
+
+  const iterable = {
+    async next() {
+      throw theError;
+    },
+    [Symbol.asyncIterator]: () => iterable
+  };
+
+  const rs = ReadableStream.from(iterable);
+  const reader = rs.getReader();
+
+  await Promise.all([
+    promise_rejects_exactly(t, theError, reader.read()),
+    promise_rejects_exactly(t, theError, reader.closed)
+  ]);
+
+}, `ReadableStream.from: stream errors when next() rejects`);
+
+promise_test(async t => {
+
+  const iterable = {
+    next() {
+      return new Promise(() => {});
+    },
+    [Symbol.asyncIterator]: () => iterable
+  };
+
+  const rs = ReadableStream.from(iterable);
+  const reader = rs.getReader();
+
+  await Promise.race([
+    reader.read().then(t.unreached_func('read() should not resolve'), t.unreached_func('read() should not reject')),
+    reader.closed.then(t.unreached_func('closed should not resolve'), t.unreached_func('closed should not reject')),
+    flushAsyncEvents()
+  ]);
+
+}, 'ReadableStream.from: stream stalls when next() never settles');
+
+promise_test(async () => {
+
+  let nextCalls = 0;
+  let nextArgs;
+  const iterable = {
+    async next(...args) {
+      nextCalls += 1;
+      nextArgs = args;
+      return { value: 'a', done: false };
+    },
+    [Symbol.asyncIterator]: () => iterable
+  };
+
+  const rs = ReadableStream.from(iterable);
+  const reader = rs.getReader();
+
+  await flushAsyncEvents();
+  assert_equals(nextCalls, 0, 'next() should not be called yet');
+
+  const read = await reader.read();
+  assert_object_equals(read, { value: 'a', done: false }, 'first read should be correct');
+  assert_equals(nextCalls, 1, 'next() should be called after first read()');
+  assert_array_equals(nextArgs, [], 'next() should be called with no arguments');
+
+}, `ReadableStream.from: calls next() after first read()`);
+
+promise_test(async t => {
+
+  const theError = new Error('a unique string');
+
+  let returnCalls = 0;
+  let returnArgs;
+  let resolveReturn;
+  const iterable = {
+    next: t.unreached_func('next() should not be called'),
+    throw: t.unreached_func('throw() should not be called'),
+    async return(...args) {
+      returnCalls += 1;
+      returnArgs = args;
+      await new Promise(r => resolveReturn = r);
+      return { done: true };
+    },
+    [Symbol.asyncIterator]: () => iterable
+  };
+
+  const rs = ReadableStream.from(iterable);
+  const reader = rs.getReader();
+  assert_equals(returnCalls, 0, 'return() should not be called yet');
+
+  let cancelResolved = false;
+  const cancelPromise = reader.cancel(theError).then(() => {
+    cancelResolved = true;
+  });
+
+  await flushAsyncEvents();
+  assert_equals(returnCalls, 1, 'return() should be called');
+  assert_array_equals(returnArgs, [theError], 'return() should be called with cancel reason');
+  assert_false(cancelResolved, 'cancel() should not resolve while promise from return() is pending');
+
+  resolveReturn();
+  await Promise.all([
+    cancelPromise,
+    reader.closed
+  ]);
+
+}, `ReadableStream.from: cancelling the returned stream calls and awaits return()`);
+
+promise_test(async t => {
+
+  let nextCalls = 0;
+  let returnCalls = 0;
+
+  const iterable = {
+    async next() {
+      nextCalls += 1;
+      return { value: undefined, done: true };
+    },
+    throw: t.unreached_func('throw() should not be called'),
+    async return() {
+      returnCalls += 1;
+    },
+    [Symbol.asyncIterator]: () => iterable
+  };
+
+  const rs = ReadableStream.from(iterable);
+  const reader = rs.getReader();
+
+  const read = await reader.read();
+  assert_object_equals(read, { value: undefined, done: true }, 'first read should be done');
+  assert_equals(nextCalls, 1, 'next() should be called once');
+
+  await reader.closed;
+  assert_equals(returnCalls, 0, 'return() should not be called');
+
+}, `ReadableStream.from: return() is not called when iterator completes normally`);
+
+promise_test(async t => {
+
+  const theError = new Error('a unique string');
+
+  const iterable = {
+    next: t.unreached_func('next() should not be called'),
+    throw: t.unreached_func('throw() should not be called'),
+    async return() {
+      return 42;
+    },
+    [Symbol.asyncIterator]: () => iterable
+  };
+
+  const rs = ReadableStream.from(iterable);
+  const reader = rs.getReader();
+
+  await promise_rejects_js(t, TypeError, reader.cancel(theError), 'cancel() should reject with a TypeError');
+
+  await reader.closed;
+
+}, `ReadableStream.from: cancel() rejects when return() fulfills with a non-object`);
+
+promise_test(async () => {
+
+  let nextCalls = 0;
+  let reader;
+  let values = ['a', 'b', 'c'];
+
+  const iterable = {
+    async next() {
+      nextCalls += 1;
+      if (nextCalls === 1) {
+        reader.read();
+      }
+      return { value: values.shift(), done: false };
+    },
+    [Symbol.asyncIterator]: () => iterable
+  };
+
+  const rs = ReadableStream.from(iterable);
+  reader = rs.getReader();
+
+  const read1 = await reader.read();
+  assert_object_equals(read1, { value: 'a', done: false }, 'first read should be correct');
+  await flushAsyncEvents();
+  assert_equals(nextCalls, 2, 'next() should be called two times');
+
+  const read2 = await reader.read();
+  assert_object_equals(read2, { value: 'c', done: false }, 'second read should be correct');
+  assert_equals(nextCalls, 3, 'next() should be called three times');
+
+}, `ReadableStream.from: reader.read() inside next()`);
+
+promise_test(async () => {
+
+  let nextCalls = 0;
+  let returnCalls = 0;
+  let reader;
+
+  const iterable = {
+    async next() {
+      nextCalls++;
+      await reader.cancel();
+      assert_equals(returnCalls, 1, 'return() should be called once');
+      return { value: 'something else', done: false };
+    },
+    async return() {
+      returnCalls++;
+    },
+    [Symbol.asyncIterator]: () => iterable
+  };
+
+  const rs = ReadableStream.from(iterable);
+  reader = rs.getReader();
+
+  const read = await reader.read();
+  assert_object_equals(read, { value: undefined, done: true }, 'first read should be done');
+  assert_equals(nextCalls, 1, 'next() should be called once');
+
+  await reader.closed;
+
+}, `ReadableStream.from: reader.cancel() inside next()`);
+
+promise_test(async t => {
+
+  let returnCalls = 0;
+  let reader;
+
+  const iterable = {
+    next: t.unreached_func('next() should not be called'),
+    async return() {
+      returnCalls++;
+      await reader.cancel();
+      return { done: true };
+    },
+    [Symbol.asyncIterator]: () => iterable
+  };
+
+  const rs = ReadableStream.from(iterable);
+  reader = rs.getReader();
+
+  await reader.cancel();
+  assert_equals(returnCalls, 1, 'return() should be called once');
+
+  await reader.closed;
+
+}, `ReadableStream.from: reader.cancel() inside return()`);
+
+promise_test(async t => {
+
+  let array = ['a', 'b'];
+
+  const rs = ReadableStream.from(array);
+  const reader = rs.getReader();
+
+  const read1 = await reader.read();
+  assert_object_equals(read1, { value: 'a', done: false }, 'first read should be correct');
+  const read2 = await reader.read();
+  assert_object_equals(read2, { value: 'b', done: false }, 'second read should be correct');
+
+  array.push('c');
+
+  const read3 = await reader.read();
+  assert_object_equals(read3, { value: 'c', done: false }, 'third read after push() should be correct');
+  const read4 = await reader.read();
+  assert_object_equals(read4, { value: undefined, done: true }, 'fourth read should be done');
+
+  await reader.closed;
+
+}, `ReadableStream.from(array), push() to array while reading`);
diff --git a/test/fixtures/wpt/streams/readable-streams/owning-type-message-port.any.js b/test/fixtures/wpt/streams/readable-streams/owning-type-message-port.any.js
new file mode 100644
index 00000000000000..e9961ce042256a
--- /dev/null
+++ b/test/fixtures/wpt/streams/readable-streams/owning-type-message-port.any.js
@@ -0,0 +1,49 @@
+// META: global=window,worker
+// META: script=../resources/test-utils.js
+// META: script=../resources/rs-utils.js
+'use strict';
+
+promise_test(async () => {
+  const channel = new MessageChannel;
+  const port1 = channel.port1;
+  const port2 = channel.port2;
+
+  const source = {
+    start(controller) {
+      controller.enqueue(port1, { transfer : [ port1 ] });
+    },
+    type: 'owning'
+  };
+
+  const stream = new ReadableStream(source);
+
+  const chunk = await stream.getReader().read();
+
+  assert_not_equals(chunk.value, port1);
+
+  let promise = new Promise(resolve => port2.onmessage = e => resolve(e.data));
+  chunk.value.postMessage("toPort2");
+  assert_equals(await promise, "toPort2");
+
+  promise = new Promise(resolve => chunk.value.onmessage = e => resolve(e.data));
+  port2.postMessage("toPort1");
+  assert_equals(await promise, "toPort1");
+}, 'Transferred MessageChannel works as expected');
+
+promise_test(async t => {
+  const channel = new MessageChannel;
+  const port1 = channel.port1;
+  const port2 = channel.port2;
+
+  const source = {
+    start(controller) {
+      controller.enqueue({ port1 }, { transfer : [ port1 ] });
+    },
+    type: 'owning'
+  };
+
+  const stream = new ReadableStream(source);
+  const [clone1, clone2] = stream.tee();
+
+  await promise_rejects_dom(t, "DataCloneError", clone2.getReader().read());
+}, 'Second branch of owning ReadableStream tee should end up into errors with transfer only values');
diff --git a/test/fixtures/wpt/streams/readable-streams/owning-type-video-frame.any.js b/test/fixtures/wpt/streams/readable-streams/owning-type-video-frame.any.js
new file mode 100644
index 00000000000000..ec01fda0b3c737
--- /dev/null
+++ b/test/fixtures/wpt/streams/readable-streams/owning-type-video-frame.any.js
@@ -0,0 +1,128 @@
+// META: global=window,worker
+// META: script=../resources/test-utils.js
+// META: script=../resources/rs-utils.js
+'use strict';
+
+function createVideoFrame()
+{
+  let init = {
+    format: 'I420',
+    timestamp: 1234,
+    codedWidth: 4,
+    codedHeight: 2
+  };
+  let data = new Uint8Array([
+    1, 2, 3, 4, 5, 6, 7, 8,  // y
+    1, 2,                    // u
+    1, 2,                    // v
+  ]);
+
+  return new VideoFrame(data, init);
+}
+
+promise_test(async () => {
+  const videoFrame = createVideoFrame();
+  videoFrame.test = 1;
+  const source = {
+    start(controller) {
+      assert_equals(videoFrame.format, 'I420');
+      controller.enqueue(videoFrame, { transfer : [ videoFrame ] });
+      assert_equals(videoFrame.format, null);
+      assert_equals(videoFrame.test, 1);
+    },
+    type: 'owning'
+  };
+
+  const stream = new ReadableStream(source);
+  // Cancelling the stream should close all video frames, thus no console messages of GCing VideoFrames should happen.
+  stream.cancel();
+}, 'ReadableStream of type owning should close serialized chunks');
+
+promise_test(async () => {
+  const videoFrame = createVideoFrame();
+  videoFrame.test = 1;
+  const source = {
+    start(controller) {
+      assert_equals(videoFrame.format, 'I420');
+      controller.enqueue({ videoFrame }, { transfer : [ videoFrame ] });
+      assert_equals(videoFrame.format, null);
+      assert_equals(videoFrame.test, 1);
+    },
+    type: 'owning'
+  };
+
+  const stream = new ReadableStream(source);
+  const reader = stream.getReader();
+
+  const chunk = await reader.read();
+  assert_equals(chunk.value.videoFrame.format, 'I420');
+  assert_equals(chunk.value.videoFrame.test, undefined);
+
+  chunk.value.videoFrame.close();
+}, 'ReadableStream of type owning should transfer JS chunks with transferred values');
+
+promise_test(async t => {
+  const videoFrame = createVideoFrame();
+  videoFrame.close();
+  const source = {
+    start(controller) {
+      assert_throws_dom("DataCloneError", () => controller.enqueue(videoFrame, { transfer : [ videoFrame ] }));
+    },
+    type: 'owning'
+  };
+
+  const stream = new ReadableStream(source);
+  const reader = stream.getReader();
+
+  await promise_rejects_dom(t, "DataCloneError", reader.read());
+}, 'ReadableStream of type owning should error when trying to enqueue not serializable values');
+
+promise_test(async () => {
+  const videoFrame = createVideoFrame();
+  const source = {
+    start(controller) {
+      controller.enqueue(videoFrame, { transfer : [ videoFrame ] });
+    },
+    type: 'owning'
+  };
+
+  const stream = new ReadableStream(source);
+  const [clone1, clone2] = stream.tee();
+
+  const chunk1 = await clone1.getReader().read();
+  const chunk2 = await clone2.getReader().read();
+
+  assert_equals(videoFrame.format, null);
+  assert_equals(chunk1.value.format, 'I420');
+  assert_equals(chunk2.value.format, 'I420');
+
+  chunk1.value.close();
+  chunk2.value.close();
+}, 'ReadableStream of type owning should clone serializable objects when teeing');
+
+promise_test(async () => {
+  const videoFrame = createVideoFrame();
+  videoFrame.test = 1;
+  const source = {
+    start(controller) {
+      assert_equals(videoFrame.format, 'I420');
+      controller.enqueue({ videoFrame }, { transfer : [ videoFrame ] });
+      assert_equals(videoFrame.format, null);
+      assert_equals(videoFrame.test, 1);
+    },
+    type: 'owning'
+  };
+
+  const stream = new ReadableStream(source);
+  const [clone1, clone2] = stream.tee();
+
+  const chunk1 = await clone1.getReader().read();
+  const chunk2 = await clone2.getReader().read();
+
+  assert_equals(videoFrame.format, null);
+  assert_equals(chunk1.value.videoFrame.format, 'I420');
+  assert_equals(chunk2.value.videoFrame.format, 'I420');
+
+  chunk1.value.videoFrame.close();
+  chunk2.value.videoFrame.close();
+}, 'ReadableStream of type owning should clone JS Objects with serializables when teeing');
diff --git a/test/fixtures/wpt/streams/readable-streams/owning-type.any.js b/test/fixtures/wpt/streams/readable-streams/owning-type.any.js
new file mode 100644
index 00000000000000..27a3dda894e4d6
--- /dev/null
+++ b/test/fixtures/wpt/streams/readable-streams/owning-type.any.js
@@ -0,0 +1,91 @@
+// META: global=window,worker
+// META: script=../resources/test-utils.js
+// META: script=../resources/rs-utils.js
+'use strict';
+
+test(() => {
+  new ReadableStream({ type: 'owning' }); // ReadableStream constructed with 'owning' type
+}, 'ReadableStream can be constructed with owning type');
+
+test(() => {
+  let startCalled = false;
+
+  const source = {
+    start(controller) {
+      assert_equals(this, source, 'source is this during start');
+      assert_true(controller instanceof ReadableStreamDefaultController, 'default controller');
+      startCalled = true;
+    },
+    type: 'owning'
+  };
+
+  new ReadableStream(source);
+  assert_true(startCalled);
+}, 'ReadableStream of type owning should call start with a ReadableStreamDefaultController');
+
+test(() => {
+  let startCalled = false;
+
+  const source = {
+    start(controller) {
+      controller.enqueue("a", { transfer: [] });
+      controller.enqueue("a", { transfer: undefined });
+      startCalled = true;
+    },
+    type: 'owning'
+  };
+
+  new ReadableStream(source);
+  assert_true(startCalled);
+}, 'ReadableStream should be able to call enqueue with an empty transfer list');
+
+test(() => {
+  let startCalled = false;
+
+  const uint8Array = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8]);
+  const buffer = uint8Array.buffer;
+  let source = {
+    start(controller) {
+      startCalled = true;
+      assert_throws_js(TypeError, () => { controller.enqueue(buffer, { transfer : [ buffer ] }); }, "transfer list is not empty");
+    }
+  };
+
+  new ReadableStream(source);
+  assert_true(startCalled);
+
+  startCalled = false;
+  source = {
+    start(controller) {
+      startCalled = true;
+      assert_throws_js(TypeError, () => { controller.enqueue(buffer, { get transfer() { throw new TypeError(); } }) }, "getter throws");
+    }
+  };
+
+  new ReadableStream(source);
+  assert_true(startCalled);
+}, 'ReadableStream should check transfer parameter');
+
+promise_test(async () => {
+  const uint8Array = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8]);
+  const buffer = uint8Array.buffer;
+  buffer.test = 1;
+  const source = {
+    start(controller) {
+      assert_equals(buffer.byteLength, 8);
+      controller.enqueue(buffer, { transfer : [ buffer ] });
+      assert_equals(buffer.byteLength, 0);
+      assert_equals(buffer.test, 1);
+    },
+    type: 'owning'
+  };
+
+  const stream = new ReadableStream(source);
+  const reader = stream.getReader();
+
+  const chunk = await reader.read();
+
+  assert_not_equals(chunk.value, buffer);
+  assert_equals(chunk.value.byteLength, 8);
+  assert_equals(chunk.value.test, undefined);
+}, 'ReadableStream of type owning should transfer enqueued chunks');
diff --git a/test/fixtures/wpt/streams/readable-streams/read-task-handling.window.js b/test/fixtures/wpt/streams/readable-streams/read-task-handling.window.js
new file mode 100644
index 00000000000000..2edc0ddddfe692
--- /dev/null
+++ b/test/fixtures/wpt/streams/readable-streams/read-task-handling.window.js
@@ -0,0 +1,46 @@
+// META: global=window,worker
+'use strict';
+
+function performMicrotaskCheckpoint() {
+    document.createNodeIterator(document, -1, {
+        acceptNode() {
+            return NodeFilter.FILTER_ACCEPT;
+        }
+    }).nextNode();
+}
+
+test(() => {
+    // Add a getter for "then" that will incidentally be invoked
+    // during promise resolution.
+    Object.prototype.__defineGetter__('then', () => {
+        // Clean up behind ourselves.
+        delete Object.prototype.then;
+
+        // This promise should (like all promises) be resolved
+        // asynchronously.
+        var executed = false;
+        Promise.resolve().then(_ => { executed = true; });
+
+        // This shouldn't run microtasks!  They should only run
+        // after the fetch is resolved.
+        performMicrotaskCheckpoint();
+
+        // The fulfill handler above shouldn't have run yet.  If it has run,
+        // throw to reject this promise and fail the test.
+        assert_false(executed, "shouldn't have run microtasks yet");
+
+        // Otherwise act as if there's no "then" property so the promise
+        // fulfills and the test passes.
+        return undefined;
+    });
+
+    const readable = new ReadableStream({
+        pull(c) {
+            c.enqueue({});
+        }
+    }, { highWaterMark: 0 });
+
+    // Create a read request, incidentally resolving a promise with an
+    // object value, thereby invoking the getter installed above.
+    readable.getReader().read();
+}, "reading from a stream should occur in a microtask scope");
diff --git a/test/fixtures/wpt/streams/transferable/transform-stream-members.any.js b/test/fixtures/wpt/streams/transferable/transform-stream-members.any.js
new file mode 100644
index 00000000000000..fca060b0c05fa3
--- /dev/null
+++ b/test/fixtures/wpt/streams/transferable/transform-stream-members.any.js
@@ -0,0 +1,16 @@
+const combinations = [
+  (t => [t, t.readable])(new TransformStream()),
+  (t => [t.readable, t])(new TransformStream()),
+  (t => [t, t.writable])(new TransformStream()),
+  (t => [t.writable, t])(new TransformStream()),
+];
+
+for (const combination of combinations) {
+  test(() => {
+    assert_throws_dom(
+      "DataCloneError",
+      () => structuredClone(combination, { transfer: combination }),
+      "structuredClone should throw"
+    );
+  }, `Transferring ${combination} should fail`);
+}
diff --git a/test/fixtures/wpt/streams/transform-streams/invalid-realm.tentative.window.js b/test/fixtures/wpt/streams/transform-streams/invalid-realm.tentative.window.js
new file mode 100644
index 00000000000000..57cdfd94862dfe
--- /dev/null
+++ b/test/fixtures/wpt/streams/transform-streams/invalid-realm.tentative.window.js
@@ -0,0 +1,17 @@
+// TransformStream should still work even if the realm is detached.
+
+// Adds an iframe to the document and returns it.
+function addIframe() {
+  const iframe = document.createElement('iframe');
+  document.body.appendChild(iframe);
+  return iframe;
+}
+
+promise_test(async t => {
+  const iframe = addIframe();
+  const stream = new iframe.contentWindow.TransformStream();
+  const readPromise = stream.readable.getReader().read();
+  const writer = stream.writable.getWriter();
+  iframe.remove();
+  return Promise.all([writer.write('A'), readPromise]);
+}, 'TransformStream: write in detached realm should succeed');
diff --git a/test/fixtures/wpt/versions.json b/test/fixtures/wpt/versions.json
index 46b4ad412c8f97..21d93af89f03d8 100644
--- a/test/fixtures/wpt/versions.json
+++ b/test/fixtures/wpt/versions.json
@@ -64,7 +64,7 @@
     "path": "resources"
   },
   "streams": {
-    "commit": "51750bc8d749bd80930506f603940a6bafda459f",
+    "commit": "517e945bbfaf903f37a35c11700eb96662efbdd3",
     "path": "streams"
   },
   "url": {
diff --git a/test/wpt/status/streams.json b/test/wpt/status/streams.json
index 187b1758741785..11e225e5f3213e 100644
--- a/test/wpt/status/streams.json
+++ b/test/wpt/status/streams.json
@@ -6,16 +6,59 @@
       ]
     }
   },
+  "piping/general-addition.any.js": {
+    "fail": {
+      "expected": [
+        "enqueue() must not synchronously call write algorithm"
+      ]
+    }
+  },
   "queuing-strategies-size-function-per-global.window.js": {
     "skip": "Browser-specific test"
   },
   "readable-streams/cross-realm-crash.window.js": {
     "skip": "Browser-specific test"
   },
+  "readable-streams/owning-type-message-port.any.js": {
+    "fail": {
+      "expected": [
+        "Transferred MessageChannel works as expected",
+        "Second branch of owning ReadableStream tee should end up into errors with transfer only values"
+      ]
+    }
+  },
+  "readable-streams/owning-type-video-frame.any.js": {
+    "skip": "Browser-specific test"
+  },
+  "readable-streams/owning-type.any.js": {
+    "fail": {
+      "expected": [
+        "ReadableStream can be constructed with owning type",
+        "ReadableStream of type owning should call start with a ReadableStreamDefaultController",
+        "ReadableStream should be able to call enqueue with an empty transfer list",
+        "ReadableStream should check transfer parameter",
+        "ReadableStream of type owning should transfer enqueued chunks"
+      ]
+    }
+  },
   "transferable/deserialize-error.window.js": {
     "skip": "Browser-specific test"
   },
   "transferable/transfer-with-messageport.window.js": {
     "skip": "Browser-specific test"
+  },
+  "transferable/transform-stream-members.any.js": {
+    "fail": {
+      "expected": [
+        "Transferring [object TransformStream],[object ReadableStream] should fail",
+        "Transferring [object TransformStream],[object WritableStream] should fail"
+      ]
+    }
+  },
+  "transform-streams/invalid-realm.tentative.window.js": {
+    "skip": "Browser-specific test"
+  },
+  "readable-streams/read-task-handling.window.js": {
+    "skip": "Browser-specific test"
   }
 }