proof of concept

Author: yaacovCR

packages/delegate/src/defaultMergedResolver.ts

@@ -1,6 +1,6 @@
 import { defaultFieldResolver, GraphQLResolveInfo } from 'graphql';
 
-import { getResponseKeyFromInfo } from '@graphql-tools/utils';
+import { getResponseKeyFromInfo, ExecutionResult } from '@graphql-tools/utils';
 
 import { resolveExternalValue } from './resolveExternalValue';
 import { getSubschema } from './Subschema';
@@ -18,7 +18,7 @@ export function defaultMergedResolver(
   args: Record<string, any>,
   context: Record<string, any>,
   info: GraphQLResolveInfo
-) {
+): any {
   if (!parent) {
     return null;
   }
@@ -34,16 +34,26 @@ export function defaultMergedResolver(
   const data = parent[responseKey];
   const unpathedErrors = getUnpathedErrors(parent);
 
-  // To Do: extract this section to separate function.
-  // If proxied result with defer or stream, result may be initially undefined
-  // so must call out to a to-be-created Receiver abstraction
-  // (as opposed to Dispatcher within graphql-js) that will notify of data arrival
-
-  if (data === undefined) {
-    return 'test';
+  // To Do:
+  // add support for transforms
+  // call out to Receiver abstraction that will publish all patches with channel based on path
+  // edit code below to subscribe to appropriate channel based on path
+  // so as to handle multiple defer patches and discriminate between them without need for labels
+
+  if (data === undefined && 'ASYNC_ITERABLE' in parent) {
+    const asyncIterable = parent['ASYNC_ITERABLE'];
+    return asyncIterableToResult(asyncIterable).then(patch => {
+      return defaultMergedResolver(patch.data, args, context, info);
+    });
   }
 
   const subschema = getSubschema(parent, responseKey);
 
   return resolveExternalValue(data, unpathedErrors, subschema, context, info);
 }
+
+async function asyncIterableToResult(asyncIterable: AsyncIterable<ExecutionResult>): Promise<any> {
+  const asyncIterator = asyncIterable[Symbol.asyncIterator]();
+  const payload = await asyncIterator.next();
+  return payload.value;
+}

packages/delegate/src/delegateToSchema.ts

@@ -1,6 +1,5 @@
 import {
   subscribe,
-  execute,
   validate,
   GraphQLSchema,
   isSchema,
@@ -13,9 +12,11 @@ import {
   GraphQLObjectType,
 } from 'graphql';
 
+import { execute } from 'graphql/experimental';
+
 import isPromise from 'is-promise';
 
-import { mapAsyncIterator, ExecutionResult } from '@graphql-tools/utils';
+import { mapAsyncIterator, ExecutionResult, isAsyncIterable } from '@graphql-tools/utils';
 
 import {
   IDelegateToSchemaOptions,
@@ -25,6 +26,7 @@ import {
   StitchingInfo,
   Endpoint,
   Transform,
+  Executor,
 } from './types';
 
 import { isSubschemaConfig } from './Subschema';
@@ -187,8 +189,16 @@ export function delegateRequest({
       info,
     });
 
-    if (isPromise(executionResult)) {
-      return executionResult.then(originalResult => transformer.transformResult(originalResult));
+    if (isAsyncIterable(executionResult)) {
+      return asyncIterableToResult(executionResult).then(originalResult => {
+        const transformedResult = transformer.transformResult(originalResult);
+        transformedResult['ASYNC_ITERABLE'] = executionResult;
+        return transformedResult;
+      });
+    } else if (isPromise(executionResult)) {
+      return (executionResult as Promise<ExecutionResult>).then(originalResult =>
+        transformer.transformResult(originalResult)
+      );
     }
     return transformer.transformResult(executionResult);
   }
@@ -201,7 +211,7 @@ export function delegateRequest({
     context,
     info,
   }).then((subscriptionResult: AsyncIterableIterator<ExecutionResult> | ExecutionResult) => {
-    if (Symbol.asyncIterator in subscriptionResult) {
+    if (isAsyncIterable(subscriptionResult)) {
       // "subscribe" to the subscription result and map the result through the transforms
       return mapAsyncIterator<ExecutionResult, any>(
         subscriptionResult as AsyncIterableIterator<ExecutionResult>,
@@ -227,15 +237,15 @@ function validateRequest(targetSchema: GraphQLSchema, document: DocumentNode) {
   }
 }
 
-function createDefaultExecutor(schema: GraphQLSchema, rootValue: Record<string, any>) {
-  return ({ document, context, variables, info }: ExecutionParams) =>
+function createDefaultExecutor(schema: GraphQLSchema, rootValue: Record<string, any>): Executor {
+  return (({ document, context, variables, info }: ExecutionParams) =>
     execute({
       schema,
       document,
       contextValue: context,
       variableValues: variables,
       rootValue: rootValue ?? info?.rootValue,
-    });
+    })) as Executor;
 }
 
 function createDefaultSubscriber(schema: GraphQLSchema, rootValue: Record<string, any>) {
@@ -248,3 +258,9 @@ function createDefaultSubscriber(schema: GraphQLSchema, rootValue: Record<string
       rootValue: rootValue ?? info?.rootValue,
     }) as any;
 }
+
+async function asyncIterableToResult(asyncIterable: AsyncIterable<ExecutionResult>): Promise<any> {
+  const asyncIterator = asyncIterable[Symbol.asyncIterator]();
+  const payload = await asyncIterator.next();
+  return payload.value;
+}

packages/delegate/src/getBatchingExecutor.ts

@@ -4,17 +4,17 @@ import isPromise from 'is-promise';
 
 import DataLoader from 'dataloader';
 
-import { ExecutionResult } from '@graphql-tools/utils';
+import { ExecutionResult, isAsyncIterable } from '@graphql-tools/utils';
 
-import { ExecutionParams, Endpoint } from './types';
+import { ExecutionParams, Endpoint, Executor } from './types';
 import { memoize2of3 } from './memoize';
 import { mergeExecutionParams } from './mergeExecutionParams';
 import { splitResult } from './splitResult';
 
 export const getBatchingExecutor = memoize2of3(function (
   _context: Record<string, any>,
   endpoint: Endpoint,
-  executor: ({ document, context, variables, info }: ExecutionParams) => ExecutionResult | Promise<ExecutionResult>
+  executor: Executor
 ) {
   const loader = new DataLoader(
     createLoadFn(
@@ -27,7 +27,7 @@ export const getBatchingExecutor = memoize2of3(function (
 });
 
 function createLoadFn(
-  executor: ({ document, context, variables, info }: ExecutionParams) => ExecutionResult | Promise<ExecutionResult>,
+  executor: Executor,
   extensionsReducer: (mergedExtensions: Record<string, any>, executionParams: ExecutionParams) => Record<string, any>
 ) {
   return async (execs: Array<ExecutionParams>): Promise<Array<ExecutionResult>> => {
@@ -53,10 +53,16 @@ function createLoadFn(
       const mergedExecutionParams = mergeExecutionParams(execBatch, extensionsReducer);
       const executionResult = executor(mergedExecutionParams);
 
-      if (isPromise(executionResult)) {
+      if (isAsyncIterable(executionResult)) {
+        throw new Error('batching not yet possible with queries that return an async iterable (defer/stream)');
+        // requires splitting up the async iterable into multiple async iterables by path versus possibly just promises
+        // so requires analyzing which of the results would get an async iterable (ie included defer/stream within the subdocument)
+        // or returning an async iterable even though defer/stream was not actually present, which is probably simpler
+        // but most probably against the spec.
+      } else if (isPromise(executionResult)) {
         containsPromises = true;
       }
-      executionResults.push(executionResult);
+      executionResults.push(executionResult as ExecutionResult);
     });
 
     if (containsPromises) {

packages/delegate/src/types.ts

@@ -14,6 +14,8 @@ import {
   GraphQLError,
 } from 'graphql';
 
+import { AsyncExecutionResult } from 'graphql/experimental';
+
 import DataLoader from 'dataloader';
 
 import { Operation, Request, TypeMap, ExecutionResult } from '@graphql-tools/utils';
@@ -131,7 +133,9 @@ export type AsyncExecutor = <
 export type SyncExecutor = <TReturn = Record<string, any>, TArgs = Record<string, any>, TContext = Record<string, any>>(
   params: ExecutionParams<TArgs, TContext>
 ) => ExecutionResult<TReturn>;
-export type Executor = AsyncExecutor | SyncExecutor;
+export type Executor = <TReturn = Record<string, any>, TArgs = Record<string, any>, TContext = Record<string, any>>(
+  params: ExecutionParams<TArgs, TContext>
+) => Promise<ExecutionResult<TReturn>> | ExecutionResult<TReturn> | AsyncIterable<AsyncExecutionResult>;
 export type Subscriber = <TReturn = Record<string, any>, TArgs = Record<string, any>, TContext = Record<string, any>>(
   params: ExecutionParams<TArgs, TContext>
 ) => Promise<AsyncIterator<ExecutionResult<TReturn>> | ExecutionResult<TReturn>>;

packages/delegate/tests/deferStream.test.ts

@@ -1,35 +1,41 @@
 import { graphql } from 'graphql/experimental';
 
 import { makeExecutableSchema } from '@graphql-tools/schema';
+import { stitchSchemas } from '@graphql-tools/stitch';
+import { isAsyncIterable } from '@graphql-tools/utils';
 
 describe('defer support', () => {
-  test('should work', async () => {
+  test('should work for root fields', async () => {
     const schema = makeExecutableSchema({
       typeDefs: `
         type Query {
-          test(input: String): String
+          test: String
         }
       `,
       resolvers: {
         Query: {
-          test: (_root, args) => args.input,
+          test: () => 'test',
         }
       },
     });
 
+    const stitchedSchema = stitchSchemas({
+      subschemas: [schema]
+    });
+
     const result = await graphql(
-      schema,
+      stitchedSchema,
       `
         query {
           ... on Query @defer {
-            test(input: "test")
+            test
           }
         }
       `,
     );
 
     const results = [];
-    if (result[Symbol.asyncIterator]) {
+    if (isAsyncIterable(result)) {
       for await (let patch of result) {
         results.push(patch);
       }
@@ -46,6 +52,63 @@ describe('defer support', () => {
       hasNext: false,
       path: [],
     });
+  });
 
+  test('should work for nested fields', async () => {
+    const schema = makeExecutableSchema({
+      typeDefs: `
+        type Object {
+          test: String
+        }
+        type Query {
+          object: Object
+        }
+      `,
+      resolvers: {
+        Object: {
+          test: () => 'test',
+        },
+        Query: {
+          object: () => ({}),
+        }
+      },
+    });
+
+    const stitchedSchema = stitchSchemas({
+      subschemas: [schema]
+    });
+
+    const result = await graphql(
+      stitchedSchema,
+      `
+        query {
+          object {
+            ... on Object @defer {
+              test
+            }
+          }
+        }
+      `,
+    );
+
+    const results = [];
+    if (isAsyncIterable(result)) {
+      for await (let patch of result) {
+        results.push(patch);
+      }
+    }
+
+    expect(results[0]).toEqual({
+      data: { object: {} },
+      hasNext: true,
+    });
+    expect(results[1]).toEqual({
+      data: {
+        test: 'test'
+      },
+      hasNext: false,
+      path: ['object'],
+    });
   });
 });
+