Support returning async iterables from resolver functions

Author: robrichard

src/__tests__/starWarsIntrospection-test.js

@@ -36,6 +36,7 @@ describe('Star Wars Introspection Tests', () => {
             { name: 'Character' },
             { name: 'String' },
             { name: 'Episode' },
+            { name: 'Int' },
             { name: 'Droid' },
             { name: 'Query' },
             { name: 'Boolean' },

src/__tests__/starWarsQuery-test.js

@@ -86,6 +86,47 @@ describe('Star Wars Query Tests', () => {
     });
   });
 
+  describe('Async Fields', () => {
+    it('Allows us to query lists that are resolved by async iterators', async () => {
+      const source = `
+        query AsyncIterableQuery {
+          human(id: "1003") {
+            friendsAsync {
+              id
+              name
+            }
+          }
+        }
+      `;
+
+      const result = await graphql({ schema, source });
+      expect(result).to.deep.equal({
+        data: {
+          human: {
+            friendsAsync: [
+              {
+                id: '1000',
+                name: 'Luke Skywalker',
+              },
+              {
+                id: '1002',
+                name: 'Han Solo',
+              },
+              {
+                id: '2000',
+                name: 'C-3PO',
+              },
+              {
+                id: '2001',
+                name: 'R2-D2',
+              },
+            ],
+          },
+        },
+      });
+    });
+  });
+
   describe('Nested Queries', () => {
     it('Allows us to query for the friends of friends of R2-D2', async () => {
       const source = `
@@ -515,5 +556,48 @@ describe('Star Wars Query Tests', () => {
         ],
       });
     });
+
+    it('Correctly reports errors raised in an async iterator', async () => {
+      const source = `
+        query HumanFriendsQuery {
+          human(id: "1003") {
+            friendsAsync(errorIndex: 2) {
+              id
+              name
+            }
+          }
+        }
+      `;
+
+      const result = await graphql({ schema, source });
+      expect(result).to.deep.equal({
+        errors: [
+          {
+            message: 'uh oh',
+            locations: [
+              {
+                line: 4,
+                column: 13,
+              },
+            ],
+            path: ['human', 'friendsAsync', 2],
+          },
+        ],
+        data: {
+          human: {
+            friendsAsync: [
+              {
+                id: '1000',
+                name: 'Luke Skywalker',
+              },
+              {
+                id: '1002',
+                name: 'Han Solo',
+              },
+            ],
+          },
+        },
+      });
+    });
   });
 });

src/__tests__/starWarsSchema.js

@@ -3,7 +3,7 @@
 import invariant from '../jsutils/invariant';
 
 import { GraphQLSchema } from '../type/schema';
-import { GraphQLString } from '../type/scalars';
+import { GraphQLString, GraphQLInt } from '../type/scalars';
 import {
   GraphQLList,
   GraphQLNonNull,
@@ -170,6 +170,29 @@ const humanType = new GraphQLObjectType({
         'The friends of the human, or an empty list if they have none.',
       resolve: (human) => getFriends(human),
     },
+    friendsAsync: {
+      type: GraphQLList(characterInterface),
+      description:
+        'The friends of the droid, or an empty list if they have none. Returns an AsyncIterable',
+      args: {
+        errorIndex: { type: GraphQLInt },
+      },
+      async *resolve(droid, { errorIndex }) {
+        const friends = getFriends(droid);
+        let i = 0;
+        for (const friend of friends) {
+          // eslint-disable-next-line no-await-in-loop
+          await new Promise((r) => setTimeout(r, 1));
+          if (i === errorIndex) {
+            throw new Error('uh oh');
+          }
+          yield friend;
+          i++;
+        }
+        // close iterator asynchronously
+        await new Promise((r) => setTimeout(r, 1));
+      },
+    },
     appearsIn: {
       type: GraphQLList(episodeEnum),
       description: 'Which movies they appear in.',

src/execution/execute.js

@@ -1,6 +1,7 @@
 // @flow strict
 
 import arrayFrom from '../polyfills/arrayFrom';
+import { SYMBOL_ASYNC_ITERATOR } from '../polyfills/symbols';
 
 import type { Path } from '../jsutils/Path';
 import type { ObjMap } from '../jsutils/ObjMap';
@@ -10,6 +11,7 @@ import memoize3 from '../jsutils/memoize3';
 import invariant from '../jsutils/invariant';
 import devAssert from '../jsutils/devAssert';
 import isPromise from '../jsutils/isPromise';
+import isAsyncIterable from '../jsutils/isAsyncIterable';
 import isObjectLike from '../jsutils/isObjectLike';
 import isCollection from '../jsutils/isCollection';
 import promiseReduce from '../jsutils/promiseReduce';
@@ -916,6 +918,56 @@ function completeValue(
   );
 }
 
+/**
+ * Complete a async iterable value by completing each item in the list with
+ * the inner type
+ */
+
+function completeAsyncIterableValue(
+  exeContext: ExecutionContext,
+  returnType: GraphQLList<GraphQLOutputType>,
+  fieldNodes: $ReadOnlyArray<FieldNode>,
+  info: GraphQLResolveInfo,
+  path: Path,
+  result: AsyncIterable<mixed>,
+): Promise<$ReadOnlyArray<mixed>> {
+  // $FlowFixMe
+  const iteratorMethod = result[SYMBOL_ASYNC_ITERATOR];
+  const iterator = iteratorMethod.call(result);
+
+  const completedResults = [];
+  let index = 0;
+
+  const itemType = returnType.ofType;
+
+  const handleNext = () => {
+    const fieldPath = addPath(path, index);
+    return iterator.next().then(
+      ({ value, done }) => {
+        if (done) {
+          return;
+        }
+        completedResults.push(
+          completeValue(
+            exeContext,
+            itemType,
+            fieldNodes,
+            info,
+            fieldPath,
+            value,
+          ),
+        );
+        index++;
+        return handleNext();
+      },
+      (error) =>
+        handleFieldError(error, fieldNodes, fieldPath, itemType, exeContext),
+    );
+  };
+
+  return handleNext().then(() => completedResults);
+}
+
 /**
  * Complete a list value by completing each item in the list with the
  * inner type
@@ -928,6 +980,17 @@ function completeListValue(
   path: Path,
   result: mixed,
 ): PromiseOrValue<$ReadOnlyArray<mixed>> {
+  if (isAsyncIterable(result)) {
+    return completeAsyncIterableValue(
+      exeContext,
+      returnType,
+      fieldNodes,
+      info,
+      path,
+      result,
+    );
+  }
+
   if (!isCollection(result)) {
     throw new GraphQLError(
       `Expected Iterable, but did not find one for field "${info.parentType.name}.${info.fieldName}".`,

src/jsutils/isAsyncIterable.js

@@ -0,0 +1,19 @@
+// @flow strict
+
+import { SYMBOL_ASYNC_ITERATOR } from '../polyfills/symbols';
+
+/**
+ * Returns true if the provided object implements the AsyncIterator protocol via
+ * either implementing a `Symbol.asyncIterator` or `"@@asyncIterator"` method.
+ */
+declare function isAsyncIterable(value: mixed): boolean %checks(value instanceof
+  AsyncIterable);
+
+// eslint-disable-next-line no-redeclare
+export default function isAsyncIterable(maybeAsyncIterable) {
+  if (maybeAsyncIterable == null || typeof maybeAsyncIterable !== 'object') {
+    return false;
+  }
+
+  return typeof maybeAsyncIterable[SYMBOL_ASYNC_ITERATOR] === 'function';
+}

src/subscription/subscribe.js

@@ -1,8 +1,7 @@
 // @flow strict
 
-import { SYMBOL_ASYNC_ITERATOR } from '../polyfills/symbols';
-
 import inspect from '../jsutils/inspect';
+import isAsyncIterable from '../jsutils/isAsyncIterable';
 import { addPath, pathToArray } from '../jsutils/Path';
 
 import { GraphQLError } from '../error/GraphQLError';
@@ -160,7 +159,7 @@ function subscribeImpl(
     // Note: Flow can't refine isAsyncIterable, so explicit casts are used.
     isAsyncIterable(resultOrStream)
       ? mapAsyncIterator(
-          ((resultOrStream: any): AsyncIterable<mixed>),
+          resultOrStream,
           mapSourceToResponse,
           reportGraphQLError,
         )
@@ -280,8 +279,7 @@ export function createSourceEventStream(
 
       // Assert field returned an event stream, otherwise yield an error.
       if (isAsyncIterable(eventStream)) {
-        // Note: isAsyncIterable above ensures this will be correct.
-        return ((eventStream: any): AsyncIterable<mixed>);
+        return eventStream;
       }
 
       throw new Error(
@@ -298,15 +296,3 @@ export function createSourceEventStream(
       : Promise.reject(error);
   }
 }
-
-/**
- * Returns true if the provided object implements the AsyncIterator protocol via
- * either implementing a `Symbol.asyncIterator` or `"@@asyncIterator"` method.
- */
-function isAsyncIterable(maybeAsyncIterable: mixed): boolean {
-  if (maybeAsyncIterable == null || typeof maybeAsyncIterable !== 'object') {
-    return false;
-  }
-
-  return typeof maybeAsyncIterable[SYMBOL_ASYNC_ITERATOR] === 'function';
-}