fixes test failures

Author: yaacovCR

src/execution/__tests__/stream-test.ts

@@ -1344,25 +1344,19 @@ describe('Execute: stream directive', () => {
         } /* c8 ignore stop */,
       },
     });
-    expectJSON(result).toDeepEqual([
-      {
-        errors: [
-          {
-            message:
-              'Cannot return null for non-nullable field NestedObject.nonNullScalarField.',
-            locations: [{ line: 4, column: 11 }],
-            path: ['nestedObject', 'nonNullScalarField'],
-          },
-        ],
-        data: {
-          nestedObject: null,
+    expectJSON(result).toDeepEqual({
+      errors: [
+        {
+          message:
+            'Cannot return null for non-nullable field NestedObject.nonNullScalarField.',
+          locations: [{ line: 4, column: 11 }],
+          path: ['nestedObject', 'nonNullScalarField'],
         },
-        hasNext: true,
-      },
-      {
-        hasNext: false,
+      ],
+      data: {
+        nestedObject: null,
       },
-    ]);
+    });
   });
   it('Filters payloads that are nulled by a later synchronous error', async () => {
     const document = parse(`
@@ -1503,9 +1497,6 @@ describe('Execute: stream directive', () => {
             ],
           },
         ],
-        hasNext: true,
-      },
-      {
         hasNext: false,
       },
     ]);

src/execution/execute.ts

@@ -122,6 +122,7 @@ export interface ExecutionContext {
   subscribeFieldResolver: GraphQLFieldResolver<any, any>;
   errors: Array<GraphQLError>;
   subsequentPayloads: Set<AsyncPayloadRecord>;
+  streams: Set<StreamContext>;
 }
 
 /**
@@ -504,6 +505,7 @@ export function buildExecutionContext(
     typeResolver: typeResolver ?? defaultTypeResolver,
     subscribeFieldResolver: subscribeFieldResolver ?? defaultFieldResolver,
     subsequentPayloads: new Set(),
+    streams: new Set(),
     errors: [],
   };
 }
@@ -516,6 +518,7 @@ function buildPerEventExecutionContext(
     ...exeContext,
     rootValue: payload,
     subsequentPayloads: new Set(),
+    streams: new Set(),
     errors: [],
   };
 }
@@ -1036,6 +1039,8 @@ async function completeAsyncIteratorValue(
       typeof stream.initialCount === 'number' &&
       index >= stream.initialCount
     ) {
+      const streamContext: StreamContext = { path: pathToArray(path) };
+      exeContext.streams.add(streamContext);
       // eslint-disable-next-line @typescript-eslint/no-floating-promises
       executeStreamIterator(
         index,
@@ -1045,6 +1050,7 @@ async function completeAsyncIteratorValue(
         info,
         itemType,
         path,
+        streamContext,
         stream.label,
         asyncPayloadRecord,
       );
@@ -1129,6 +1135,7 @@ function completeListValue(
   let previousAsyncPayloadRecord = asyncPayloadRecord;
   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.
@@ -1139,6 +1146,8 @@ function completeListValue(
       typeof stream.initialCount === 'number' &&
       index >= stream.initialCount
     ) {
+      streamContext = { path: pathToArray(path) };
+      exeContext.streams.add(streamContext);
       previousAsyncPayloadRecord = executeStreamField(
         path,
         itemPath,
@@ -1147,6 +1156,7 @@ function completeListValue(
         fieldNodes,
         info,
         itemType,
+        streamContext,
         stream.label,
         previousAsyncPayloadRecord,
       );
@@ -1173,6 +1183,10 @@ function completeListValue(
     index++;
   }
 
+  if (streamContext) {
+    exeContext.streams.delete(streamContext);
+  }
+
   return containsPromise ? Promise.all(completedResults) : completedResults;
 }
 
@@ -1813,12 +1827,14 @@ function executeStreamField(
   fieldNodes: ReadonlyArray<FieldNode>,
   info: GraphQLResolveInfo,
   itemType: GraphQLOutputType,
+  streamContext: StreamContext,
   label?: string,
   parentContext?: AsyncPayloadRecord,
 ): AsyncPayloadRecord {
   const asyncPayloadRecord = new StreamRecord({
     label,
     path: itemPath,
+    streamContext,
     parentContext,
     exeContext,
   });
@@ -1965,19 +1981,20 @@ async function executeStreamIterator(
   info: GraphQLResolveInfo,
   itemType: GraphQLOutputType,
   path: Path,
+  streamContext: StreamContext,
   label?: string,
   parentContext?: AsyncPayloadRecord,
 ): Promise<void> {
   let index = initialIndex;
   let previousAsyncPayloadRecord = parentContext ?? undefined;
-  // eslint-disable-next-line no-constant-condition
-  while (true) {
+  while (exeContext.streams.has(streamContext)) {
     const itemPath = addPath(path, index, undefined);
     const asyncPayloadRecord = new StreamRecord({
       label,
       path: itemPath,
       parentContext: previousAsyncPayloadRecord,
       iterator,
+      streamContext,
       exeContext,
     });
 
@@ -2003,6 +2020,7 @@ async function executeStreamIterator(
           // ignore errors
         });
       }
+      exeContext.streams.delete(streamContext);
       return;
     }
 
@@ -2025,6 +2043,7 @@ async function executeStreamIterator(
     asyncPayloadRecord.addItems(completedItems);
 
     if (done) {
+      exeContext.streams.delete(streamContext);
       break;
     }
     previousAsyncPayloadRecord = asyncPayloadRecord;
@@ -2038,6 +2057,15 @@ function filterSubsequentPayloads(
   currentAsyncRecord: AsyncPayloadRecord | undefined,
 ): void {
   const nullPathArray = pathToArray(nullPath);
+  exeContext.streams.forEach((stream) => {
+    for (let i = 0; i < nullPathArray.length; i++) {
+      if (stream.path[i] !== nullPathArray[i]) {
+        // stream points to a path unaffected by this payload
+        return;
+      }
+    }
+    exeContext.streams.delete(stream);
+  });
   exeContext.subsequentPayloads.forEach((asyncRecord) => {
     if (asyncRecord === currentAsyncRecord) {
       // don't remove payload from where error originates
@@ -2055,7 +2083,7 @@ function filterSubsequentPayloads(
         // ignore error
       });
     }
-    exeContext.subsequentPayloads.delete(asyncRecord);
+exeContext.subsequentPayloads.delete(asyncRecord);
   });
 }
 
@@ -2211,6 +2239,9 @@ class DeferredFragmentRecord {
     this._resolve?.(data);
   }
 }
+interface StreamContext {
+  path: Array<string | number>;
+}
 
 class StreamRecord {
   type: 'stream';
@@ -2221,6 +2252,7 @@ class StreamRecord {
   promise: Promise<void>;
   parentContext: AsyncPayloadRecord | undefined;
   iterator: AsyncIterator<unknown> | undefined;
+  streamContext: StreamContext;
   isCompletedIterator?: boolean;
   isCompleted: boolean;
   _exeContext: ExecutionContext;
@@ -2229,6 +2261,7 @@ class StreamRecord {
     label: string | undefined;
     path: Path | undefined;
     iterator?: AsyncIterator<unknown>;
+    streamContext: StreamContext;
     parentContext: AsyncPayloadRecord | undefined;
     exeContext: ExecutionContext;
   }) {
@@ -2238,6 +2271,7 @@ class StreamRecord {
     this.path = pathToArray(opts.path);
     this.parentContext = opts.parentContext;
     this.iterator = opts.iterator;
+    this.streamContext = opts.streamContext;
     this.errors = [];
     this._exeContext = opts.exeContext;
     this._exeContext.subsequentPayloads.add(this);