replace mapAsyncIterator with repeater-based mapAsyncIterable

Author: yaacovCR

src/execution/__tests__/mapAsyncIterable-test.ts

@@ -109,7 +109,7 @@ describe('mapAsyncIterable', () => {
 
     // Early return
     expect(await doubles.return('')).to.deep.equal({
-      value: 'The End',
+      value: '',
       done: true,
     });
 
@@ -148,20 +148,24 @@ describe('mapAsyncIterable', () => {
 
     // Early return
     expect(await doubles.return(0)).to.deep.equal({
-      value: undefined,
+      value: 0,
       done: true,
     });
   });
 
   it('passes through early return from async values', async () => {
+    let didVisitFinally = false;
+
     async function* source() {
       try {
         yield 'a';
         /* c8 ignore next 3 */
         yield 'b';
         yield 'c'; // Shouldn't be reached.
       } finally {
+        didVisitFinally = true;
         yield 'Done';
+        /* c8 ignore next 2 */
         yield 'Last';
       }
     }
@@ -173,19 +177,21 @@ describe('mapAsyncIterable', () => {
 
     // Early return
     expect(await doubles.return()).to.deep.equal({
-      value: 'DoneDone',
-      done: false,
+      value: undefined,
+      done: true,
     });
 
-    // Subsequent next calls may yield from finally block
+    // Subsequent next calls
     expect(await doubles.next()).to.deep.equal({
-      value: 'LastLast',
-      done: false,
+      value: undefined,
+      done: true,
     });
     expect(await doubles.next()).to.deep.equal({
       value: undefined,
       done: true,
     });
+
+    expect(didVisitFinally).to.equal(true);
   });
 
   it('allows throwing errors through async iterable', async () => {
@@ -277,6 +283,7 @@ describe('mapAsyncIterable', () => {
       } finally {
         didVisitFinally = true;
         yield 1000;
+        /* c8 ignore next 1 */
       }
     }
 

src/execution/mapAsyncIterable.ts

@@ -1,57 +1,73 @@
+import { isPromise } from '../jsutils/isPromise';
 import type { PromiseOrValue } from '../jsutils/PromiseOrValue';
+import type { Push, Stop } from '../jsutils/Repeater';
+import { Repeater } from '../jsutils/Repeater';
 
 /**
- * Given an AsyncIterable and a callback function, return an AsyncIterator
+ * Given an AsyncIterable and a callback function, return an AsyncGenerator
  * which produces values mapped via calling the callback function.
  */
 export function mapAsyncIterable<T, U, R = undefined>(
   iterable: AsyncGenerator<T, R, void> | AsyncIterable<T>,
-  callback: (value: T) => PromiseOrValue<U>,
+  fn: (value: T) => PromiseOrValue<U>,
 ): AsyncGenerator<U, R, void> {
-  const iterator = iterable[Symbol.asyncIterator]();
+  return new Repeater<U, R, void>(async (push, stop) => {
+    const iter: AsyncIterator<T, R, void> = iterable[Symbol.asyncIterator]();
+    let finalIteration: PromiseOrValue<IteratorResult<T, R>> | undefined;
+    // The stop promise never rejects.
+    // eslint-disable-next-line @typescript-eslint/no-floating-promises
+    stop.then(() => {
+      if (typeof iter.return === 'function') {
+        finalIteration = iter.return();
+      }
+    });
 
-  async function mapResult(
-    result: IteratorResult<T, R>,
-  ): Promise<IteratorResult<U, R>> {
-    if (result.done) {
-      return result;
+    // eslint-disable-next-line no-unmodified-loop-condition
+    while (!finalIteration) {
+      // eslint-disable-next-line no-await-in-loop
+      if (await handleNext(iter, iter.next(), fn, push, stop)) {
+        break;
+      }
     }
 
-    try {
-      return { value: await callback(result.value), done: false };
-    } catch (error) {
-      /* c8 ignore start */
-      // FIXME: add test case
-      if (typeof iterator.return === 'function') {
-        try {
-          await iterator.return();
-        } catch (_e) {
-          /* ignore error */
-        }
-      }
-      throw error;
-      /* c8 ignore stop */
+    if (isPromise(finalIteration)) {
+      await finalIteration;
     }
+
+    return undefined as unknown as R; // void :();
+  });
+}
+
+async function handleNext<T, U, R>(
+  iter: AsyncIterator<T, R, void>,
+  next: Promise<IteratorResult<T, R>>,
+  fn: (value: T) => PromiseOrValue<U>,
+  push: Push<U>,
+  stop: Stop,
+): Promise<boolean> {
+  const { done, value } = await next;
+
+  if (done) {
+    stop();
+    return true;
   }
 
-  return {
-    async next() {
-      return mapResult(await iterator.next());
-    },
-    async return(): Promise<IteratorResult<U, R>> {
-      // If iterator.return() does not exist, then type R must be undefined.
-      return typeof iterator.return === 'function'
-        ? mapResult(await iterator.return())
-        : { value: undefined as any, done: true };
-    },
-    async throw(error?: unknown) {
-      if (typeof iterator.throw === 'function') {
-        return mapResult(await iterator.throw(error));
-      }
-      throw error;
-    },
-    [Symbol.asyncIterator]() {
-      return this;
-    },
-  };
+  let mapped: U;
+  try {
+    mapped = await fn(value);
+  } catch (err) {
+    stop(err);
+    return true;
+  }
+
+  try {
+    await push(mapped);
+  } catch (err) {
+    if (typeof iter.throw === 'function') {
+      return handleNext(iter, iter.throw(err), fn, push, stop);
+    }
+    throw err;
+  }
+
+  return false;
 }