add ability to use perEventExecutor (#4211)

by exporting executeSubscriptionEvent()

and adding option for to provide a custom fn

addresses #894

cf. #2485 , #3071

Author: yaacovCR

src/execution/__tests__/subscribe-test.ts

@@ -21,7 +21,11 @@ import {
 import { GraphQLSchema } from '../../type/schema.js';
 
 import type { ExecutionArgs, ExecutionResult } from '../execute.js';
-import { createSourceEventStream, subscribe } from '../execute.js';
+import {
+  createSourceEventStream,
+  executeSubscriptionEvent,
+  subscribe,
+} from '../execute.js';
 
 import { SimplePubSub } from './simplePubSub.js';
 
@@ -326,6 +330,45 @@ describe('Subscription Initialization Phase', () => {
     });
   });
 
+  it('uses a custom default perEventExecutor', async () => {
+    const schema = new GraphQLSchema({
+      query: DummyQueryType,
+      subscription: new GraphQLObjectType({
+        name: 'Subscription',
+        fields: {
+          foo: { type: GraphQLString },
+        },
+      }),
+    });
+
+    async function* fooGenerator() {
+      yield { foo: 'FooValue' };
+    }
+
+    let count = 0;
+    const subscription = subscribe({
+      schema,
+      document: parse('subscription { foo }'),
+      rootValue: { foo: fooGenerator },
+      perEventExecutor: (validatedArgs) => {
+        count++;
+        return executeSubscriptionEvent(validatedArgs);
+      },
+    });
+    assert(isAsyncIterable(subscription));
+
+    expect(await subscription.next()).to.deep.equal({
+      done: false,
+      value: { data: { foo: 'FooValue' } },
+    });
+
+    expect(await subscription.next()).to.deep.equal({
+      done: true,
+      value: undefined,
+    });
+    expect(count).to.equal(1);
+  });
+
   it('should only resolve the first field of invalid multi-field', async () => {
     async function* fooGenerator() {
       yield { foo: 'FooValue' };

src/execution/execute.ts

@@ -127,6 +127,9 @@ export interface ValidatedExecutionArgs {
   fieldResolver: GraphQLFieldResolver<any, any>;
   typeResolver: GraphQLTypeResolver<any, any>;
   subscribeFieldResolver: GraphQLFieldResolver<any, any>;
+  perEventExecutor: (
+    validatedExecutionArgs: ValidatedExecutionArgs,
+  ) => PromiseOrValue<ExecutionResult>;
 }
 
 /**
@@ -208,6 +211,11 @@ export interface ExecutionArgs {
   fieldResolver?: Maybe<GraphQLFieldResolver<any, any>>;
   typeResolver?: Maybe<GraphQLTypeResolver<any, any>>;
   subscribeFieldResolver?: Maybe<GraphQLFieldResolver<any, any>>;
+  perEventExecutor?: Maybe<
+    (
+      validatedExecutionArgs: ValidatedExecutionArgs,
+    ) => PromiseOrValue<ExecutionResult>
+  >;
   /** Additional execution options. */
   options?: {
     /** Set the maximum number of errors allowed for coercing (defaults to 50). */
@@ -235,7 +243,7 @@ export function execute(args: ExecutionArgs): PromiseOrValue<ExecutionResult> {
     return { errors: validatedExecutionArgs };
   }
 
-  return executeOperation(validatedExecutionArgs);
+  return executeQueryOrMutationOrSubscriptionEvent(validatedExecutionArgs);
 }
 
 /**
@@ -253,7 +261,7 @@ export function execute(args: ExecutionArgs): PromiseOrValue<ExecutionResult> {
  * at which point we still log the error and null the parent field, which
  * in this case is the entire response.
  */
-function executeOperation(
+function executeQueryOrMutationOrSubscriptionEvent(
   validatedExecutionArgs: ValidatedExecutionArgs,
 ): PromiseOrValue<ExecutionResult> {
   const exeContext: ExecutionContext = {
@@ -352,6 +360,7 @@ export function validateExecutionArgs(
     fieldResolver,
     typeResolver,
     subscribeFieldResolver,
+    perEventExecutor,
     options,
   } = args;
 
@@ -425,6 +434,7 @@ export function validateExecutionArgs(
     fieldResolver: fieldResolver ?? defaultFieldResolver,
     typeResolver: typeResolver ?? defaultTypeResolver,
     subscribeFieldResolver: subscribeFieldResolver ?? defaultFieldResolver,
+    perEventExecutor: perEventExecutor ?? executeSubscriptionEvent,
   };
 }
 
@@ -1419,18 +1429,22 @@ function mapSourceToResponse(
   // For each payload yielded from a subscription, map it over the normal
   // GraphQL `execute` function, with `payload` as the rootValue.
   // This implements the "MapSourceToResponseEvent" algorithm described in
-  // the GraphQL specification. The `execute` function provides the
-  // "ExecuteSubscriptionEvent" algorithm, as it is nearly identical to the
-  // "ExecuteQuery" algorithm, for which `execute` is also used.
+  // the GraphQL specification..
   return mapAsyncIterable(resultOrStream, (payload: unknown) => {
     const perEventExecutionArgs: ValidatedExecutionArgs = {
       ...validatedExecutionArgs,
       rootValue: payload,
     };
-    return executeOperation(perEventExecutionArgs);
+    return validatedExecutionArgs.perEventExecutor(perEventExecutionArgs);
   });
 }
 
+export function executeSubscriptionEvent(
+  validatedExecutionArgs: ValidatedExecutionArgs,
+): PromiseOrValue<ExecutionResult> {
+  return executeQueryOrMutationOrSubscriptionEvent(validatedExecutionArgs);
+}
+
 /**
  * Implements the "CreateSourceEventStream" algorithm described in the
  * GraphQL specification, resolving the subscription source event stream.

src/execution/index.ts

@@ -3,6 +3,7 @@ export { pathToArray as responsePathAsArray } from '../jsutils/Path.js';
 export {
   createSourceEventStream,
   execute,
+  executeSubscriptionEvent,
   executeSync,
   defaultFieldResolver,
   defaultTypeResolver,

src/index.ts

@@ -324,6 +324,7 @@ export type {
 // Execute GraphQL queries.
 export {
   execute,
+  executeSubscriptionEvent,
   executeSync,
   defaultFieldResolver,
   defaultTypeResolver,