fix(race): concurrent next calls

graphql · Mar 19, 2021 · 9680b06 · 9680b06
1 parent 24330c0
commit 9680b06
Show file tree

Hide file tree

Showing 2 changed files with 117 additions and 34 deletions.
diff --git a/src/execution/__tests__/stream-test.js b/src/execution/__tests__/stream-test.js
@@ -389,6 +389,72 @@ describe('Execute: stream directive', () => {
       },
     ]);
   });
+  it('Can stream a field that returns an async iterable', async () => {
+    const document = parse(`
+      query { 
+        asyncIterableList @stream(initialCount: 2) {
+          name
+          id
+        }
+      }
+    `);
+    const schema = new GraphQLSchema({ query });
+
+    const result = await execute({ schema, document, rootValue: {} });
+
+    const results = [];
+    if (isAsyncIterable(result)) {
+      const asyncResults = await Promise.all([
+        result.next(),
+        result.next(),
+        result.next(),
+        result.next(),
+      ]);
+      results.push(...asyncResults);
+    }
+
+    expect(results).to.deep.equal([
+      {
+        done: false,
+        value: {
+          data: {
+            asyncIterableList: [
+              {
+                name: 'Luke',
+                id: '1',
+              },
+              {
+                name: 'Han',
+                id: '2',
+              },
+            ],
+          },
+          hasNext: true,
+        },
+      },
+      {
+        done: false,
+        value: {
+          data: {
+            name: 'Leia',
+            id: '3',
+          },
+          path: ['asyncIterableList', 2],
+          hasNext: true,
+        },
+      },
+      {
+        done: false,
+        value: {
+          hasNext: false,
+        },
+      },
+      {
+        done: true,
+        value: undefined,
+      },
+    ]);
+  });
   it('Handles error thrown in async iterable before initialCount is reached', async () => {
     const document = parse(`
       query { 

diff --git a/src/execution/execute.js b/src/execution/execute.js
@@ -1774,43 +1774,60 @@ export class Dispatcher {
       });
     }
     return new Promise((resolve) => {
+      let resolved = false;
       this._subsequentPayloads.forEach((promise) => {
-        promise.then(() => {
-          // resolve with actual promise, not resolved value of promise so we can remove it from this._subsequentPayloads
-          resolve({ promise });
-        });
-      });
-    })
-      .then(({ promise }) => {
-        this._subsequentPayloads.splice(
-          this._subsequentPayloads.indexOf(promise),
-          1,
-        );
-        return promise;
-      })
-      .then(({ value, done }) => {
-        if (done && this._subsequentPayloads.length === 0) {
-          // async iterable resolver just finished and no more pending payloads
-          return {
-            value: {
-              hasNext: false,
-            },
-            done: false,
+        promise.then((payload) => {
+          if (resolved) {
+            return;
+          }
+
+          resolved = true;
+
+          if (this._subsequentPayloads.length === 0) {
+            // a different call to next has exhausted all payloads
+            resolve({ value: undefined, done: true });
+            return;
+          }
+
+          const index = this._subsequentPayloads.indexOf(promise);
+
+          if (index === -1) {
+            // a different call to next has consumed this payload
+            resolve(this._race());
+            return;
+          }
+
+          this._subsequentPayloads.splice(index, 1);
+
+          const { value, done } = payload;
+
+          if (done && this._subsequentPayloads.length === 0) {
+            // async iterable resolver just finished and no more pending payloads
+            resolve({
+              value: {
+                hasNext: false,
+              },
+              done: false,
+            });
+            return;
+          } else if (done) {
+            // async iterable resolver just finished but there are pending payloads
+            // return the next one
+            resolve(this._race());
+            return;
+          }
+
+          const returnValue: ExecutionPatchResult = {
+            ...value,
+            hasNext: this._subsequentPayloads.length > 0,
           };
-        } else if (done) {
-          // async iterable resolver just finished but there are pending payloads
-          // return the next one
-          return this._race();
-        }
-        const returnValue: ExecutionPatchResult = {
-          ...value,
-          hasNext: this._subsequentPayloads.length > 0,
-        };
-        return {
-          value: returnValue,
-          done: false,
-        };
+          resolve({
+            value: returnValue,
+            done: false,
+          });
+        });
       });
+    });
   }
 
   _next(): Promise<IteratorResult<AsyncExecutionResult, void>> {