introduce executeStreamIterator

Author: yaacovCR

src/execution/__tests__/stream-test.ts

@@ -532,11 +532,6 @@ describe('Execute: stream directive', () => {
               },
             ],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ name: 'Leia', id: '3' }],
             path: ['friendList', 2],
@@ -978,11 +973,6 @@ describe('Execute: stream directive', () => {
               },
             ],
           },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             items: [{ nonNullName: 'Han' }],
             path: ['friendList', 2],

src/execution/execute.ts

@@ -1473,38 +1473,39 @@ function completeListValue(
   // This is specified as a simple map, however we're optimizing the path
   // where the list contains no Promises by avoiding creating another Promise.
   let containsPromise = false;
-  let currentParents = parentRecords;
   const completedResults: Array<unknown> = [];
   let index = 0;
   let streamContext: StreamContext | undefined;
-  for (const item of result) {
-    // No need to modify the info object containing the path,
-    // since from here on it is not ever accessed by resolver functions.
-    const itemPath = addPath(path, index, fieldGroup);
-
+  const iterator = result[Symbol.iterator]();
+  // eslint-disable-next-line no-constant-condition
+  while (true) {
     if (streamUsage && index >= streamUsage.initialCount) {
-      if (streamContext === undefined) {
-        streamContext = {
-          label: streamUsage.label,
-          path: pathToArray(path),
-        };
-      }
-      currentParents = executeStreamField(
-        path,
-        itemPath,
-        item,
+      streamContext = {
+        label: streamUsage.label,
+        path: pathToArray(path),
+      };
+      executeStreamIterator(
+        index,
+        iterator,
         exeContext,
+        itemType,
         getStreamedFieldGroup(fieldGroup, streamUsage),
         info,
-        itemType,
-        streamContext,
+        path,
         deferMap,
-        currentParents,
+        streamContext,
+        parentRecords,
       );
-      index++;
-      continue;
+      break;
+    }
+
+    const iteration = iterator.next();
+    if (iteration.done) {
+      break;
     }
 
+    const item = iteration.value;
+    const itemPath = addPath(path, index, fieldGroup);
     addPendingDeferredField(fieldGroup, itemPath, deferMap);
     if (
       completeListItemValue(
@@ -2171,86 +2172,71 @@ function assertEventStream(result: unknown): AsyncIterable<unknown> {
   return result;
 }
 
-function executeStreamField(
-  path: Path<FieldGroup>,
-  itemPath: Path<FieldGroup>,
-  item: PromiseOrValue<unknown>,
+function executeStreamIterator(
+  initialIndex: number,
+  iterator: Iterator<unknown>,
   exeContext: ExecutionContext,
+  itemType: GraphQLOutputType,
   fieldGroup: FieldGroup,
   info: GraphQLResolveInfo,
-  itemType: GraphQLOutputType,
-  streamContext: StreamContext,
+  path: Path<FieldGroup>,
   deferMap: Map<DeferUsage, DeferredFragmentRecord>,
-  parents?: Array<AsyncPayloadRecord> | undefined,
-): Array<StreamRecord> {
-  const streamRecord = new StreamRecord({
-    streamContext,
-    path: itemPath,
-    parents,
-    exeContext,
-  });
-  const currentParents = [streamRecord];
-  if (isPromise(item)) {
-    const completedItems = completePromisedValue(
-      exeContext,
-      itemType,
-      fieldGroup,
-      info,
-      itemPath,
-      item,
-      deferMap,
-      streamRecord,
-      currentParents,
-    ).then(
-      (value) => [value],
-      (error) => {
-        streamRecord.errors.push(error);
-        filterSubsequentPayloads(exeContext, path, currentParents);
-        return null;
-      },
-    );
+  streamContext: StreamContext,
+  parents: Array<AsyncPayloadRecord> | undefined,
+): void {
+  let index = initialIndex;
+  let currentParents = parents;
+  // eslint-disable-next-line no-constant-condition
+  while (true) {
+    const iteration = iterator.next();
+    if (iteration.done) {
+      break;
+    }
 
-    streamRecord.addItems(completedItems);
-    return currentParents;
-  }
+    const item = iteration.value;
+    const itemPath = addPath(path, index, fieldGroup);
+    const streamRecord = new StreamRecord({
+      streamContext,
+      path: itemPath,
+      parents: currentParents,
+      exeContext,
+    });
 
-  let completedItem: PromiseOrValue<unknown>;
-  try {
-    try {
-      completedItem = completeValue(
+    currentParents = [streamRecord];
+    if (isPromise(item)) {
+      // eslint-disable-next-line @typescript-eslint/no-floating-promises
+      handlePromisedItemForStream(
         exeContext,
         itemType,
         fieldGroup,
         info,
+        path,
         itemPath,
         item,
         deferMap,
         streamRecord,
         currentParents,
       );
-    } catch (rawError) {
-      handleFieldError(
-        rawError,
-        exeContext,
-        itemType,
-        fieldGroup,
-        itemPath,
-        deferMap,
-        streamRecord,
-      );
-      completedItem = null;
-      filterSubsequentPayloads(exeContext, itemPath, currentParents);
+
+      index++;
+      continue;
     }
-  } catch (error) {
-    streamRecord.errors.push(error);
-    filterSubsequentPayloads(exeContext, path, currentParents);
-    streamRecord.addItems(null);
-    return currentParents;
-  }
 
-  if (isPromise(completedItem)) {
-    const completedItems = completedItem
-      .then(undefined, (rawError) => {
+    let completedItem: PromiseOrValue<unknown>;
+    try {
+      try {
+        completedItem = completeValue(
+          exeContext,
+          itemType,
+          fieldGroup,
+          info,
+          itemPath,
+          item,
+          deferMap,
+          streamRecord,
+          currentParents,
+        );
+      } catch (rawError) {
         handleFieldError(
           rawError,
           exeContext,
@@ -2261,23 +2247,91 @@ function executeStreamField(
           streamRecord,
         );
         filterSubsequentPayloads(exeContext, itemPath, currentParents);
-        return null;
-      })
-      .then(
-        (value) => [value],
-        (error) => {
-          streamRecord.errors.push(error);
-          filterSubsequentPayloads(exeContext, path, currentParents);
-          return null;
-        },
+        completedItem = null;
+      }
+    } catch (error) {
+      if (fieldGroup.inInitialResult) {
+        streamRecord.errors.push(error);
+      }
+      returnStreamIteratorIgnoringError(streamContext);
+      filterSubsequentPayloads(exeContext, path, currentParents);
+      streamRecord.addItems(null);
+      break;
+    }
+
+    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]);
+    }
 
-    streamRecord.addItems(completedItems);
-    return currentParents;
+    index++;
   }
+}
 
-  streamRecord.addItems([completedItem]);
-  return currentParents;
+async function handlePromisedItemForStream(
+  exeContext: ExecutionContext,
+  returnType: GraphQLOutputType,
+  fieldGroup: FieldGroup,
+  info: GraphQLResolveInfo,
+  path: Path<FieldGroup>,
+  itemPath: Path<FieldGroup>,
+  item: Promise<unknown>,
+  deferMap: Map<DeferUsage, DeferredFragmentRecord>,
+  streamRecord: StreamRecord,
+  parentRecords: Array<AsyncPayloadRecord> | undefined,
+): Promise<void> {
+  try {
+    try {
+      const resolvedItem = await item;
+      let completedItem = completeValue(
+        exeContext,
+        returnType,
+        fieldGroup,
+        info,
+        itemPath,
+        resolvedItem,
+        deferMap,
+        streamRecord,
+        parentRecords,
+      );
+      // TODO: add test for this
+      /* c8 ignore next 3 */
+      if (isPromise(completedItem)) {
+        completedItem = await completedItem;
+      }
+      streamRecord.addItems([completedItem]);
+    } catch (rawError) {
+      handleFieldError(
+        rawError,
+        exeContext,
+        returnType,
+        fieldGroup,
+        itemPath,
+        deferMap,
+        streamRecord,
+      );
+      filterSubsequentPayloads(exeContext, itemPath, [streamRecord]);
+      streamRecord.addItems([null]);
+    }
+  } catch (error) {
+    if (fieldGroup.inInitialResult) {
+      streamRecord.errors.push(error);
+    }
+    filterSubsequentPayloads(exeContext, path, [streamRecord]);
+    streamRecord.addItems(null);
+  }
 }
 
 async function executeStreamAsyncIterator(
@@ -2765,7 +2819,7 @@ class StreamRecord {
     });
   }
 
-  addItems(items: PromiseOrValue<Array<unknown> | null>) {
+  addItems(items: Array<unknown> | null) {
     if (this.parents !== undefined) {
       const parentPromises = this.parents.map((parent) => parent.promise);
       this._resolve?.(Promise.any(parentPromises).then(() => items));