Use async_sequence in ReadableStream.from

index.bs

@@ -484,7 +484,7 @@ The Web IDL definition for the {{ReadableStream}} class is given as follows:
 interface ReadableStream {
   constructor(optional object underlyingSource, optional QueuingStrategy strategy = {});
 
-  static ReadableStream from(any asyncIterable);
+  static ReadableStream from(async_sequence<any> asyncIterable);
 
   readonly attribute boolean locked;
 
@@ -2170,39 +2170,20 @@ The following abstract operations operate on {{ReadableStream}} instances at a h
  ReadableStreamFromIterable(|asyncIterable|)</dfn> performs the following steps:
 
  1. Let |stream| be undefined.
- 1. Let |iteratorRecord| be ? [$GetIterator$](|asyncIterable|, async).
+ 1. Let |iterator| be the result of [=opening an async sequence=] |asyncIterable|.
  1. Let |startAlgorithm| be an algorithm that returns undefined.
  1. Let |pullAlgorithm| be the following steps:
-  1. Let |nextResult| be [$IteratorNext$](|iteratorRecord|).
-  1. If |nextResult| is an abrupt completion, return [=a promise rejected with=]
-     |nextResult|.\[[Value]].
-  1. Let |nextPromise| be [=a promise resolved with=] |nextResult|.\[[Value]].
-  1. Return the result of [=reacting=] to |nextPromise| with the following fulfillment steps,
-     given |iterResult|:
-   1. If |iterResult| [=is not an Object=], throw a {{TypeError}}.
-   1. Let |done| be ? [$IteratorComplete$](|iterResult|).
-   1. If |done| is true:
-    1. Perform ! [$ReadableStreamDefaultControllerClose$](|stream|.[=ReadableStream/[[controller]]=]).
-   1. Otherwise:
-    1. Let |value| be ? [$IteratorValue$](|iterResult|).
-    1. Perform ! [$ReadableStreamDefaultControllerEnqueue$](|stream|.[=ReadableStream/[[controller]]=],
-       |value|).
+  1. Let |nextPromise| be the result of [=get an async iterator next value|getting the next value=]
+     of |iterator|.
+  1. Return the result of [=reacting=] to |nextPromise|:
+   - If |nextPromise| was fulfilled with value |v|:
+    1. If |v| is [=end of iteration=], perform ! [$ReadableStreamDefaultControllerClose$](|stream|.[=ReadableStream/[[controller]]=]).
        <!-- TODO (future): If we allow changing the queuing strategy, this Enqueue might throw.
-             We'll then need to catch the error and close the async iterator. -->
+          We'll then need to catch the error and close the async iterator. -->
+    1. Otherwise, perform ! [$ReadableStreamDefaultControllerEnqueue$](|stream|.[=ReadableStream/[[controller]]=], |v|).
+   - If |nextPromise| was rejected with reason |r|, perform ! [$ReadableStreamDefaultControllerError$](|stream|.[=ReadableStream/[[controller]]=], |r|).
  1. Let |cancelAlgorithm| be the following steps, given |reason|:
-  1. Let |iterator| be |iteratorRecord|.\[[Iterator]].
-  1. Let |returnMethod| be [$GetMethod$](|iterator|, "`return`").
-  1. If |returnMethod| is an abrupt completion, return [=a promise rejected with=]
-     |returnMethod|.\[[Value]].
-  1. If |returnMethod|.\[[Value]] is undefined, return [=a promise resolved with=] undefined.
-  1. Let |returnResult| be [$Call$](|returnMethod|.\[[Value]], |iterator|, « |reason| »).
-  1. If |returnResult| is an abrupt completion, return [=a promise rejected with=]
-     |returnResult|.\[[Value]].
-  1. Let |returnPromise| be [=a promise resolved with=] |returnResult|.\[[Value]].
-  1. Return the result of [=reacting=] to |returnPromise| with the following fulfillment steps,
-     given |iterResult|:
-   1. If |iterResult| [=is not an Object=], throw a {{TypeError}}.
-   1. Return undefined.
+  1. Return the result of [=close an async iterator|closing=] |iterator| with |reason|.
  1. Set |stream| to ! [$CreateReadableStream$](|startAlgorithm|, |pullAlgorithm|, |cancelAlgorithm|,
     0).
  1. Return |stream|.
@@ -6986,6 +6967,13 @@ to grow organically as needed.
 [=ReadableStream/set up with byte reading support=] operations directly on the [=this=] value inside
 their constructor steps.
 
+<div algorithm="create a ReadableStream from an async sequence">
+ To <dfn export lt="create from async sequence|create from an async sequence" for="ReadableStream">create</dfn>
+ a {{ReadableStream}} from an [=async sequence=] |sequence|:
+
+ 1. Return ? [$ReadableStreamFromIterable$](|sequence|).
+</div>
+
 <hr>
 
 The following algorithms must only be used on {{ReadableStream}} instances initialized via the above

reference-implementation/lib/ReadableStream.webidl

@@ -2,7 +2,7 @@
 interface ReadableStream {
   constructor(optional object underlyingSource, optional QueuingStrategy strategy = {});
 
-  static ReadableStream from(any asyncIterable);
+  static ReadableStream from(async_sequence<any> asyncIterable);
 
   readonly attribute boolean locked;
 

reference-implementation/lib/abstract-ops/ecmascript.js

@@ -29,64 +29,6 @@ exports.IsDetachedBuffer = O => {
   return ArrayBufferPrototypeDetachedGetter.call(O) === true;
 };
 
-exports.Call = (F, V, args = []) => {
-  if (typeof F !== 'function') {
-    throw new TypeError('Argument is not a function');
-  }
-
-  return Reflect.apply(F, V, args);
-};
-
-exports.GetMethod = (V, P) => {
-  const func = V[P];
-  if (func === undefined || func === null) {
-    return undefined;
-  }
-  if (typeof func !== 'function') {
-    throw new TypeError(`${P} is not a function`);
-  }
-  return func;
-};
-
-exports.CreateAsyncFromSyncIterator = syncIteratorRecord => {
-  // Instead of re-implementing CreateAsyncFromSyncIterator and %AsyncFromSyncIteratorPrototype%,
-  // we use yield* inside an async generator function to achieve the same result.
-
-  // Wrap the sync iterator inside a sync iterable, so we can use it with yield*.
-  const syncIterable = {
-    [Symbol.iterator]: () => syncIteratorRecord.iterator
-  };
-  // Create an async generator function and immediately invoke it.
-  const asyncIterator = (async function* () {
-    return yield* syncIterable;
-  }());
-  // Return as an async iterator record.
-  const nextMethod = asyncIterator.next;
-  return { iterator: asyncIterator, nextMethod, done: false };
-};
-
-exports.GetIterator = (obj, hint = 'sync', method) => {
-  assert(hint === 'sync' || hint === 'async');
-  if (method === undefined) {
-    if (hint === 'async') {
-      method = exports.GetMethod(obj, Symbol.asyncIterator);
-      if (method === undefined) {
-        const syncMethod = exports.GetMethod(obj, Symbol.iterator);
-        const syncIteratorRecord = exports.GetIterator(obj, 'sync', syncMethod);
-        return exports.CreateAsyncFromSyncIterator(syncIteratorRecord);
-      }
-    } else {
-      method = exports.GetMethod(obj, Symbol.iterator);
-    }
-  }
-  const iterator = exports.Call(method, obj);
-  if (!exports.typeIsObject(iterator)) {
-    throw new TypeError('The iterator method must return an object');
-  }
-  const nextMethod = iterator.next;
-  return { iterator, nextMethod, done: false };
-};
-
 exports.IteratorNext = (iteratorRecord, value) => {
   let result;
   if (value === undefined) {

reference-implementation/lib/abstract-ops/readable-streams.js

@@ -4,8 +4,8 @@ const assert = require('assert');
 const { promiseResolvedWith, promiseRejectedWith, newPromise, resolvePromise, rejectPromise, uponPromise,
         setPromiseIsHandledToTrue, waitForAllPromise, transformPromiseWith, uponFulfillment, uponRejection } =
   require('../helpers/webidl.js');
-const { CanTransferArrayBuffer, Call, CopyDataBlockBytes, CreateArrayFromList, GetIterator, GetMethod, IsDetachedBuffer,
-        IteratorComplete, IteratorNext, IteratorValue, TransferArrayBuffer, typeIsObject } = require('./ecmascript.js');
+const { CanTransferArrayBuffer, CopyDataBlockBytes, CreateArrayFromList, IsDetachedBuffer,
+        IteratorComplete, IteratorValue, TransferArrayBuffer } = require('./ecmascript.js');
 const { CanCopyDataBlockBytes, CloneAsUint8Array, IsNonNegativeNumber } = require('./miscellaneous.js');
 const { EnqueueValueWithSize, ResetQueue } = require('./queue-with-sizes.js');
 const { AcquireWritableStreamDefaultWriter, IsWritableStreamLocked, WritableStreamAbort,
@@ -1911,22 +1911,13 @@ function SetUpReadableByteStreamControllerFromUnderlyingSource(
 
 function ReadableStreamFromIterable(asyncIterable) {
   let stream;
-  const iteratorRecord = GetIterator(asyncIterable, 'async');
+  const iterator = asyncIterable[Symbol.asyncIterator]();
 
   const startAlgorithm = () => undefined;
 
   function pullAlgorithm() {
-    let nextResult;
-    try {
-      nextResult = IteratorNext(iteratorRecord);
-    } catch (e) {
-      return promiseRejectedWith(e);
-    }
-    const nextPromise = promiseResolvedWith(nextResult);
+    const nextPromise = iterator.next();
     return transformPromiseWith(nextPromise, iterResult => {
-      if (!typeIsObject(iterResult)) {
-        throw new TypeError('The promise returned by the iterator.next() method must fulfill with an object');
-      }
       const done = IteratorComplete(iterResult);
       if (done === true) {
         ReadableStreamDefaultControllerClose(stream._controller);
@@ -1938,27 +1929,8 @@ function ReadableStreamFromIterable(asyncIterable) {
   }
 
   function cancelAlgorithm(reason) {
-    const iterator = iteratorRecord.iterator;
-    let returnMethod;
-    try {
-      returnMethod = GetMethod(iterator, 'return');
-    } catch (e) {
-      return promiseRejectedWith(e);
-    }
-    if (returnMethod === undefined) {
-      return promiseResolvedWith(undefined);
-    }
-    let returnResult;
-    try {
-      returnResult = Call(returnMethod, iterator, [reason]);
-    } catch (e) {
-      return promiseRejectedWith(e);
-    }
-    const returnPromise = promiseResolvedWith(returnResult);
-    return transformPromiseWith(returnPromise, iterResult => {
-      if (!typeIsObject(iterResult)) {
-        throw new TypeError('The promise returned by the iterator.return() method must fulfill with an object');
-      }
+    const returnPromise = iterator.return(reason);
+    return transformPromiseWith(returnPromise, () => {
       return undefined;
     });
   }

reference-implementation/package.json

@@ -18,7 +18,7 @@
     "eslint": "^6.8.0",
     "minimatch": "^10.0.1",
     "opener": "^1.5.2",
-    "webidl2js": "^19.1.0",
+    "webidl2js": "^20.0.0",
     "wpt-runner": "^7.0.0"
   }
 }