fix(fusion): handle promises of async iterables

Author: ardatan

.changeset/cyan-worms-pull.md

@@ -0,0 +1,5 @@
+---
+"@graphql-mesh/fusion-execution": patch
+---
+
+Handle promises of async iterables correctly

packages/fusion/execution/src/execution.ts

@@ -306,7 +306,14 @@ export function executeResolverOperationNodesWithDependenciesInParallel({
       if (isAsyncIterable(fieldOpResult$)) {
         fieldOpAsyncIterables.push(mapAsyncIterator(fieldOpResult$, handleFieldOpResult));
       } else if (isPromise(fieldOpResult$)) {
-        fieldOpPromises.push(fieldOpResult$.then(handleFieldOpResult as any));
+        fieldOpPromises.push(
+          fieldOpResult$.then(fieldOpResult => {
+            if (isAsyncIterable(fieldOpResult)) {
+              return mapAsyncIterator(fieldOpResult, handleFieldOpResult as any);
+            }
+            return handleFieldOpResult(fieldOpResult as any);
+          }),
+        );
       } else {
         handleFieldOpResult(fieldOpResult$);
       }
@@ -373,7 +380,16 @@ export function executeResolverOperationNodesWithDependenciesInParallel({
       const mergedIterable = Repeater.merge([...fieldOpPromises, ...fieldOpAsyncIterables]);
       asyncIterables.push(mapAsyncIterator(mergedIterable, handleFieldOpResults));
     } else if (fieldOpPromises.length) {
-      dependencyPromises.push(Promise.all(fieldOpPromises).then(handleFieldOpResults));
+      dependencyPromises.push(
+        Promise.all(fieldOpPromises).then(fieldOpPromiseResults => {
+          const asyncIterablesInResults = fieldOpPromiseResults.filter(isAsyncIterable);
+          if (asyncIterablesInResults.length) {
+            const mergedIterable = Repeater.merge(asyncIterablesInResults);
+            return mapAsyncIterator(mergedIterable, handleFieldOpResults);
+          }
+          return handleFieldOpResults();
+        }),
+      );
     } else {
       handleFieldOpResults();
     }
@@ -390,7 +406,14 @@ export function executeResolverOperationNodesWithDependenciesInParallel({
     return mapAsyncIterator(mergedIterable, handleDependencyPromises);
   }
   if (dependencyPromises.length) {
-    return Promise.all(dependencyPromises).then(handleDependencyPromises);
+    return Promise.all(dependencyPromises).then(dependencyPromiseResults => {
+      const asyncIterablesInResults = dependencyPromiseResults.filter(isAsyncIterable);
+      if (asyncIterablesInResults.length) {
+        const mergedIterable = Repeater.merge(asyncIterablesInResults);
+        return mapAsyncIterator(mergedIterable, handleDependencyPromises);
+      }
+      return handleDependencyPromises();
+    });
   }
   return handleDependencyPromises();
 }
@@ -627,7 +650,12 @@ export function executeResolverOperationNode({
     }
 
     if (isPromise(result$)) {
-      return result$.then(handleResult);
+      return result$.then(result => {
+        if (isAsyncIterable(result)) {
+          return mapAsyncIterator(result as AsyncIterableIterator<any>, handleResult as any);
+        }
+        return handleResult(result);
+      });
     }
 
     return handleResult(result$);

packages/fusion/execution/src/operations.ts

@@ -295,7 +295,12 @@ export function executeOperationPlan({
     errors,
   });
   if (isPromise(res$)) {
-    return res$.then(res => prepareExecutionResult(res, errors, executablePlan));
+    return res$.then(res => {
+      if (isAsyncIterable(res)) {
+        return mapAsyncIterator(res, res => prepareExecutionResult(res, errors, executablePlan));
+      }
+      return prepareExecutionResult(res, errors, executablePlan);
+    });
   }
   if (isAsyncIterable(res$)) {
     return mapAsyncIterator(res$, res => prepareExecutionResult(res, errors, executablePlan));

packages/fusion/runtime/tests/runtime.spec.ts

@@ -38,6 +38,28 @@ describe('useFusiongraph', () => {
       },
     },
   });
+  const cSchema = createSchema({
+    typeDefs: `
+      type Query {
+        c: String
+      }
+      type Subscription {
+        c: String
+      }
+    `,
+    resolvers: {
+      Query: {
+        c: () => 'c',
+      },
+      Subscription: {
+        c: {
+          async *subscribe() {
+            yield { c: 'c' };
+          },
+        },
+      },
+    },
+  });
   const yoga = createYoga({
     plugins: [
       useFusiongraph({
@@ -51,6 +73,10 @@ describe('useFusiongraph', () => {
               name: 'b',
               schema: bSchema,
             },
+            {
+              name: 'c',
+              schema: cSchema,
+            },
           ]),
         transports() {
           return {
@@ -60,6 +86,11 @@ describe('useFusiongraph', () => {
                   return createDefaultExecutor(aSchema);
                 case 'b':
                   return createDefaultExecutor(bSchema);
+                case 'c':
+                  // Returns Promise<AsyncIterableIterator<ExecutionResult>>
+                  return async function (args) {
+                    return createDefaultExecutor(cSchema)(args);
+                  };
               }
               throw new Error(`Unknown subgraph: ${subgraphName}`);
             },
@@ -100,4 +131,28 @@ describe('useFusiongraph', () => {
       },
     });
   }
+  it('works with executors that return Promise<AsyncIterableIterator<ExecutionResult>>', async () => {
+    expect.assertions(1);
+    const res = await yoga.fetch('http://localhost:4000/graphql', {
+      method: 'POST',
+      headers: {
+        'Content-Type': 'application/json',
+        Accept: 'text/event-stream',
+      },
+      body: JSON.stringify({
+        query: `
+          subscription {
+            c
+          }
+        `,
+      }),
+    });
+    const chunks: Uint8Array[] = [];
+    // eslint-disable-next-line no-unreachable-loop
+    for await (const chunk of res.body) {
+      chunks.push(chunk);
+    }
+    const resText = Buffer.concat(chunks).toString('utf-8');
+    expect(resText).toMatch(/{"c":"c"}/);
+  });
 });