remove extra ticks from executeStreamAsyncIterator

Author: yaacovCR

src/execution/__tests__/stream-test.ts

@@ -1730,11 +1730,6 @@ describe('Execute: stream directive', () => {
             items: [{ id: '1', name: 'Luke' }],
             path: ['nestedObject', 'nestedFriendList', 0],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ id: '2', name: 'Han' }],
             path: ['nestedObject', 'nestedFriendList', 1],
@@ -1793,15 +1788,6 @@ describe('Execute: stream directive', () => {
             data: { scalarField: 'slow', nestedFriendList: [] },
             path: ['nestedObject'],
           },
-        ],
-        hasNext: true,
-      },
-      done: false,
-    });
-    const result3 = await iterator.next();
-    expectJSON(result3).toDeepEqual({
-      value: {
-        incremental: [
           {
             items: [{ name: 'Luke' }],
             path: ['nestedObject', 'nestedFriendList', 0],
@@ -1811,8 +1797,8 @@ describe('Execute: stream directive', () => {
       },
       done: false,
     });
-    const result4 = await iterator.next();
-    expectJSON(result4).toDeepEqual({
+    const result3 = await iterator.next();
+    expectJSON(result3).toDeepEqual({
       value: {
         incremental: [
           {
@@ -1824,8 +1810,8 @@ describe('Execute: stream directive', () => {
       },
       done: false,
     });
-    const result5 = await iterator.next();
-    expectJSON(result5).toDeepEqual({
+    const result4 = await iterator.next();
+    expectJSON(result4).toDeepEqual({
       value: undefined,
       done: true,
     });

src/execution/execute.ts

@@ -1347,12 +1347,12 @@ async function completeAsyncIteratorValue(
         index,
         iterator,
         exeContext,
+        itemType,
         getStreamedFieldGroup(fieldGroup, streamUsage),
         info,
-        itemType,
         path,
-        streamContext,
         deferMap,
+        streamContext,
         parentRecords,
       );
       break;
@@ -2280,99 +2280,17 @@ function executeStreamField(
   return currentParents;
 }
 
-async function executeStreamAsyncIteratorItem(
-  iterator: AsyncIterator<unknown>,
-  exeContext: ExecutionContext,
-  fieldGroup: FieldGroup,
-  info: GraphQLResolveInfo,
-  itemType: GraphQLOutputType,
-  streamRecord: StreamRecord,
-  itemPath: Path<FieldGroup>,
-  deferMap: Map<DeferUsage, DeferredFragmentRecord>,
-  parentRecords: Array<AsyncPayloadRecord>,
-): Promise<IteratorResult<unknown>> {
-  let item;
-  try {
-    const iteration = await iterator.next();
-    if (!exeContext.streams.has(streamRecord.streamContext)) {
-      streamRecord.remove();
-      return { done: true, value: undefined };
-    }
-    const { value, done } = iteration;
-    if (done) {
-      streamRecord.remove();
-      return { done, value: undefined };
-    }
-    item = value;
-  } catch (rawError) {
-    handleFieldError(
-      rawError,
-      exeContext,
-      itemType,
-      fieldGroup,
-      itemPath,
-      deferMap,
-      streamRecord,
-    );
-    // don't continue if iterator throws
-    return { done: true, value: null };
-  }
-  let completedItem;
-  try {
-    completedItem = completeValue(
-      exeContext,
-      itemType,
-      fieldGroup,
-      info,
-      itemPath,
-      item,
-      deferMap,
-      streamRecord,
-      parentRecords,
-    );
-
-    if (isPromise(completedItem)) {
-      completedItem = completedItem.then(undefined, (rawError) => {
-        handleFieldError(
-          rawError,
-          exeContext,
-          itemType,
-          fieldGroup,
-          itemPath,
-          deferMap,
-          streamRecord,
-        );
-        filterSubsequentPayloads(exeContext, itemPath, parentRecords);
-        return null;
-      });
-    }
-    return { done: false, value: completedItem };
-  } catch (rawError) {
-    handleFieldError(
-      rawError,
-      exeContext,
-      itemType,
-      fieldGroup,
-      itemPath,
-      deferMap,
-      streamRecord,
-    );
-    filterSubsequentPayloads(exeContext, itemPath, parentRecords);
-    return { done: false, value: null };
-  }
-}
-
 async function executeStreamAsyncIterator(
   initialIndex: number,
   iterator: AsyncIterator<unknown>,
   exeContext: ExecutionContext,
+  itemType: GraphQLOutputType,
   fieldGroup: FieldGroup,
   info: GraphQLResolveInfo,
-  itemType: GraphQLOutputType,
   path: Path<FieldGroup>,
-  streamContext: StreamContext,
   deferMap: Map<DeferUsage, DeferredFragmentRecord>,
-  parents?: Array<AsyncPayloadRecord> | undefined,
+  streamContext: StreamContext,
+  parents: Array<AsyncPayloadRecord> | undefined,
 ): Promise<void> {
   let index = initialIndex;
   let currentParents = parents;
@@ -2387,55 +2305,135 @@ async function executeStreamAsyncIterator(
     });
     currentParents = [streamRecord];
 
-    let iteration;
+    currentParents = [streamRecord];
+
+    let item;
     try {
       // eslint-disable-next-line no-await-in-loop
-      iteration = await executeStreamAsyncIteratorItem(
-        iterator,
+      const iteration = await iterator.next();
+      if (iteration.done) {
+        streamRecord.remove();
+        return;
+      }
+      item = iteration.value;
+    } catch (rawError) {
+      handleFieldError(
+        rawError,
         exeContext,
-        fieldGroup,
-        info,
         itemType,
-        streamRecord,
+        fieldGroup,
         itemPath,
         deferMap,
-        currentParents,
+        streamRecord,
       );
+      // don't continue if iterator throws
+      streamRecord.addItems([null]);
+      exeContext.streams.delete(streamContext);
+      return;
+    }
+
+    try {
+      let completedItem;
+      try {
+        completedItem = completeValue(
+          exeContext,
+          itemType,
+          fieldGroup,
+          info,
+          itemPath,
+          item,
+          deferMap,
+          streamRecord,
+          currentParents,
+        );
+
+        if (isPromise(completedItem)) {
+          // eslint-disable-next-line @typescript-eslint/no-floating-promises
+          handlePromisedStreamResult(
+            completedItem,
+            streamRecord,
+            exeContext,
+            itemType,
+            fieldGroup,
+            path,
+            itemPath,
+            deferMap,
+            streamContext,
+          );
+        } else {
+          streamRecord.addItems([completedItem]);
+        }
+      } catch (rawError) {
+        handleFieldError(
+          rawError,
+          exeContext,
+          itemType,
+          fieldGroup,
+          itemPath,
+          deferMap,
+          streamRecord,
+        );
+        filterSubsequentPayloads(exeContext, itemPath, currentParents);
+        streamRecord.addItems([null]);
+      }
     } catch (error) {
-      streamRecord.errors.push(error);
+      if (fieldGroup.inInitialResult) {
+        streamRecord.errors.push(error);
+      }
+      returnStreamIteratorIgnoringError(streamContext);
+      exeContext.streams.delete(streamContext);
       filterSubsequentPayloads(exeContext, path, currentParents);
       streamRecord.addItems(null);
-      // entire stream has errored and bubbled upwards
-      if (iterator?.return) {
-        iterator.return().catch(() => {
-          // ignore errors
-        });
-      }
       return;
     }
 
-    const { done, value: completedItem } = iteration;
-
-    let completedItems: PromiseOrValue<Array<unknown> | null>;
-    if (isPromise(completedItem)) {
-      completedItems = completedItem.then(
-        (value) => [value],
-        (error) => {
-          streamRecord.errors.push(error);
-          filterSubsequentPayloads(exeContext, path, [streamRecord]);
-          return null;
-        },
-      );
-    } else {
-      completedItems = [completedItem];
+    if (!exeContext.streams.has(streamContext)) {
+      // stream was filtered
+      returnStreamIteratorIgnoringError(streamContext);
+      break;
     }
 
-    streamRecord.addItems(completedItems);
+    index++;
+  }
+}
 
-    if (done) {
-      break;
+async function handlePromisedStreamResult(
+  result: Promise<unknown>,
+  streamRecord: StreamRecord,
+  exeContext: ExecutionContext,
+  itemType: GraphQLOutputType,
+  fieldGroup: FieldGroup,
+  path: Path<FieldGroup>,
+  itemPath: Path<FieldGroup>,
+  deferMap: Map<DeferUsage, DeferredFragmentRecord>,
+  streamContext: StreamContext,
+): Promise<void> {
+  try {
+    let resolved;
+    try {
+      resolved = await result;
+    } catch (rawError) {
+      handleFieldError(
+        rawError,
+        exeContext,
+        itemType,
+        fieldGroup,
+        itemPath,
+        deferMap,
+        streamRecord,
+      );
+      filterSubsequentPayloads(exeContext, itemPath, [streamRecord]);
+      resolved = null;
     }
-    index++;
+    streamRecord.addItems([resolved]);
+  } catch (error) {
+    if (fieldGroup.inInitialResult) {
+      streamRecord.errors.push(error);
+    }
+    returnStreamIteratorIgnoringError(streamContext);
+    exeContext.streams.delete(streamContext);
+    filterSubsequentPayloads(exeContext, path, [streamRecord]);
+    streamRecord.addItems(null);
   }
 }
 
@@ -2533,10 +2531,6 @@ function yieldSubsequentPayloads(
     const incremental = getCompletedIncrementalResults(exeContext);
     const hasNext = exeContext.subsequentPayloads.size > 0;
 
-    if (!incremental.length && hasNext) {
-      return next();
-    }
-
     if (!hasNext) {
       isDone = true;
     }