stream: implement ReadableStream.from

Fixes: nodejs#48389 PR-URL: nodejs#48395 Reviewed-By: Matteo Collina <[email protected]> Reviewed-By: Matthew Aitken <[email protected]> Reviewed-By: Benjamin Gruenbaum <[email protected]>
Ceres6 · Aug 14, 2023 · 0a36a9b · 0a36a9b
1 parent 6c3bfe5
commit 0a36a9b
Show file tree

Hide file tree

Showing 14 changed files with 1,036 additions and 2 deletions.
diff --git 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
@@ -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
@@ -13,13 +13,16 @@ const {
   PromiseReject,
   ReflectGet,
   Symbol,
+  SymbolAsyncIterator,
+  SymbolIterator,
   Uint8Array,
 } = primordials;
 
 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
@@ -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
@@ -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");