do not initiate multiple streams at the same path

Author: yaacovCR

src/execution/__tests__/stream-test.ts

@@ -636,6 +636,107 @@ describe('Execute: stream directive', () => {
       },
     ]);
   });
+  it('Does not initiate multiple streams at the same path', async () => {
+    const document = parse(`
+      query { 
+        friendList @stream(initialCount: 2) {
+          name
+          id
+        }
+        ... @defer {
+          friendList @stream(initialCount: 2) {
+            name
+            id
+          }
+        }
+      }
+    `);
+    const result = await complete(document, { friendList: friends });
+    expectJSON(result).toDeepEqual([
+      {
+        data: {
+          friendList: [
+            { name: 'Luke', id: '1' },
+            { name: 'Han', id: '2' },
+          ],
+        },
+        hasNext: true,
+      },
+      {
+        incremental: [
+          {
+            items: [{ name: 'Leia', id: '3' }],
+            path: ['friendList', 2],
+          },
+          {
+            data: {
+              friendList: [
+                { name: 'Luke', id: '1' },
+                { name: 'Han', id: '2' },
+              ],
+            },
+            path: [],
+          },
+        ],
+        hasNext: false,
+      },
+    ]);
+  });
+  it('Does not initiate multiple streams at the same path for async iterables', async () => {
+    const document = parse(`
+      query { 
+        friendList @stream(initialCount: 2) {
+          name
+          id
+        }
+        ... @defer {
+          friendList @stream(initialCount: 2) {
+            name
+            id
+          }
+        }
+      }
+    `);
+    const result = await complete(document, {
+      async *friendList() {
+        yield await Promise.resolve(friends[0]);
+        yield await Promise.resolve(friends[1]);
+        yield await Promise.resolve(friends[2]);
+      },
+    });
+    expectJSON(result).toDeepEqual([
+      {
+        data: {
+          friendList: [
+            { name: 'Luke', id: '1' },
+            { name: 'Han', id: '2' },
+          ],
+        },
+        hasNext: true,
+      },
+      {
+        incremental: [
+          {
+            data: {
+              friendList: [
+                { name: 'Luke', id: '1' },
+                { name: 'Han', id: '2' },
+              ],
+            },
+            path: [],
+          },
+          {
+            items: [{ name: 'Leia', id: '3' }],
+            path: ['friendList', 2],
+          },
+        ],
+        hasNext: true,
+      },
+      {
+        hasNext: false,
+      },
+    ]);
+  });
   it('Negative values of initialCount throw field errors on a field that returns an async iterable', async () => {
     const document = parse(`
       query { 

src/execution/execute.ts

@@ -125,6 +125,7 @@ export interface ExecutionContext {
   errors: Array<GraphQLError>;
   subsequentPayloads: Set<AsyncPayloadRecord>;
   branches: WeakMap<GroupedFieldSet, Set<Path | undefined>>;
+  streams: WeakMap<FieldGroup, Set<Path>>;
 }
 
 /**
@@ -505,6 +506,7 @@ export function buildExecutionContext(
     addPath: createPathFactory(),
     subsequentPayloads: new Set(),
     branches: new WeakMap(),
+    streams: new WeakMap(),
     errors: [],
   };
 }
@@ -519,6 +521,7 @@ function buildPerEventExecutionContext(
     addPath: createPathFactory(),
     subsequentPayloads: new Set(),
     branches: new WeakMap(),
+    streams: new WeakMap(),
     errors: [],
   };
 }
@@ -540,6 +543,23 @@ function shouldBranch(
   return false;
 }
 
+function shouldStream(
+  fieldGroup: FieldGroup,
+  exeContext: ExecutionContext,
+  path: Path,
+): boolean {
+  const set = exeContext.streams.get(fieldGroup);
+  if (set === undefined) {
+    exeContext.streams.set(fieldGroup, new Set([path]));
+    return true;
+  }
+  if (!set.has(path)) {
+    set.add(path);
+    return true;
+  }
+  return false;
+}
+
 /**
  * Implements the "Executing operations" section of the spec.
  */
@@ -1102,17 +1122,19 @@ async function completeAsyncIteratorValue(
       typeof stream.initialCount === 'number' &&
       index >= stream.initialCount
     ) {
-      // eslint-disable-next-line @typescript-eslint/no-floating-promises
-      executeStreamAsyncIterator(
-        index,
-        asyncIterator,
-        exeContext,
-        fieldGroup,
-        info,
-        itemType,
-        path,
-        asyncPayloadRecord,
-      );
+      if (shouldStream(fieldGroup, exeContext, path)) {
+        // eslint-disable-next-line @typescript-eslint/no-floating-promises
+        executeStreamAsyncIterator(
+          index,
+          asyncIterator,
+          exeContext,
+          fieldGroup,
+          info,
+          itemType,
+          path,
+          asyncPayloadRecord,
+        );
+      }
       break;
     }
 
@@ -1208,16 +1230,18 @@ function completeListValue(
       typeof stream.initialCount === 'number' &&
       index >= stream.initialCount
     ) {
-      executeStreamIterator(
-        index,
-        iterator,
-        exeContext,
-        fieldGroup,
-        info,
-        itemType,
-        path,
-        asyncPayloadRecord,
-      );
+      if (shouldStream(fieldGroup, exeContext, path)) {
+        executeStreamIterator(
+          index,
+          iterator,
+          exeContext,
+          fieldGroup,
+          info,
+          itemType,
+          path,
+          asyncPayloadRecord,
+        );
+      }
       break;
     }