feat: ReadableStream.from (#19446)

Closes #19417

1 files changed, 62 insertions, 0 deletions

diff --git a/ext/web/06_streams.js b/ext/web/06_streams.js
index 21207c372..beab2ec12 100644
--- a/ext/web/06_streams.js
+++ b/ext/web/06_streams.js
@@ -62,6 +62,7 @@ const {
   // SharedArrayBufferPrototype,
   Symbol,
   SymbolAsyncIterator,
+  SymbolIterator,
   SymbolFor,
   TypeError,
   TypedArrayPrototypeGetBuffer,
@@ -4780,6 +4781,30 @@ function initializeCountSizeFunction(globalObject) {
   WeakMapPrototypeSet(countSizeFunctionWeakMap, globalObject, size);
 }
 
+async function* createAsyncFromSyncIterator(syncIterator) {
+  // deno-lint-ignore prefer-primordials
+  yield* syncIterator;
+}
+
+// Ref: https://tc39.es/ecma262/#sec-getiterator
+function getIterator(obj, async = false) {
+  if (async) {
+    if (obj[SymbolAsyncIterator] === undefined) {
+      if (obj[SymbolIterator] === undefined) {
+        throw new TypeError("No iterator found");
+      }
+      return createAsyncFromSyncIterator(obj[SymbolIterator]());
+    } else {
+      return obj[SymbolAsyncIterator]();
+    }
+  } else {
+    if (obj[SymbolIterator] === undefined) {
+      throw new TypeError("No iterator found");
+    }
+    return obj[SymbolIterator]();
+  }
+}
+
 const _resourceBacking = Symbol("[[resourceBacking]]");
 // This distinction exists to prevent unrefable streams being used in
 // regular fast streams that are unaware of refability
@@ -4863,6 +4888,43 @@ class ReadableStream {
     }
   }
 
+  static from(asyncIterable) {
+    webidl.requiredArguments(
+      arguments.length,
+      1,
+      "Failed to call 'ReadableStream.from'",
+    );
+    asyncIterable = webidl.converters.any(asyncIterable);
+
+    const iterator = getIterator(asyncIterable, true);
+
+    const stream = createReadableStream(() => undefined, async () => {
+      // deno-lint-ignore prefer-primordials
+      const res = await iterator.next();
+      if (typeof res !== "object") {
+        throw new TypeError("iterator.next value is not an object");
+      }
+      if (res.done) {
+        readableStreamDefaultControllerClose(stream[_controller]);
+      } else {
+        readableStreamDefaultControllerEnqueue(stream[_controller], res.value);
+      }
+    }, async (reason) => {
+      if (typeof iterator.return === "undefined") {
+        return undefined;
+      } else {
+        // deno-lint-ignore prefer-primordials
+        const res = await iterator.return(reason);
+        if (typeof res !== "object") {
+          throw new TypeError("iterator.return value is not an object");
+        } else {
+          return undefined;
+        }
+      }
+    }, 0);
+    return stream;
+  }
+
   /** @returns {boolean} */
   get locked() {
     webidl.assertBranded(this, ReadableStreamPrototype);