Repository: graphql/graphql-js
Pull request #3636 (2022-06-10 13:37:12)

move `subscribe` code to `execute` file
= makes no other changes
= paves the way for further changes

netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *compassionate-pike-271cb3* processing.


|  Name | Link |
|---------------------------------|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | 72965246ac8f50c7e7efd1409a8950b26546d929 |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/compassionate-pike-271cb3/deploys/62a3490a8236f700085512c3 |
github-actions[bot]: Hi @yaacovCR, I'm @github-actions bot happy to help you with this PR 👋

<details>
<summary> Supported commands </summary>

Please post this commands in separate comments and only one per comment:
* `@github-actions run-benchmark` - Run benchmark comparing base and merge commits for this PR
* `@github-actions publish-pr-on-npm` - Build package from this PR and publish it on NPM
</details>

IvanGoncharov: @yaacovCR Thanks a lot for doing it as a separate step 👍 
Save a lot of effort during the review!
Changes look good once you remove the "positional test".
If you find that some popular libraries were using our "internal" functions please open a separate issue to discuss what to do and communicate with their authors.
Otherwise this PR is ready to be merge 👍 

Patch:
diff --git a/src/execution/execute.ts b/src/execution/execute.ts
index 6dc42461..67905366 100644
--- a/src/execution/execute.ts
+++ b/src/execution/execute.ts
@@ -1,6 +1,7 @@
 import { devAssert } from '../jsutils/devAssert';
 import { inspect } from '../jsutils/inspect';
 import { invariant } from '../jsutils/invariant';
+import { isAsyncIterable } from '../jsutils/isAsyncIterable';
 import { isIterableObject } from '../jsutils/isIterableObject';
 import { isObjectLike } from '../jsutils/isObjectLike';
 import { isPromise } from '../jsutils/isPromise';
@@ -51,6 +52,7 @@ import {
   collectFields,
   collectSubfields as _collectSubfields,
 } from './collectFields';
+import { mapAsyncIterator } from './mapAsyncIterator';
 import { getArgumentValues, getVariableValues } from './values';
 
 /**
@@ -235,6 +237,7 @@ function buildResponse(
  * Essential assertions before executing to provide developer feedback for
  * improper use of the GraphQL library.
  *
+ * TODO: consider no longer exporting this function
  * @internal
  */
 export function assertValidExecutionArguments(
@@ -260,6 +263,7 @@ export function assertValidExecutionArguments(
  *
  * Throws a GraphQLError if a valid execution context cannot be created.
  *
+ * TODO: consider no longer exporting this function
  * @internal
  */
 export function buildExecutionContext(
@@ -543,6 +547,7 @@ function executeField(
 }
 
 /**
+ * TODO: consider no longer exporting this function
  * @internal
  */
 export function buildResolveInfo(
@@ -1009,3 +1014,225 @@ export const defaultFieldResolver: GraphQLFieldResolver<unknown, unknown> =
       return property;
     }
   };
+
+/**
+ * Implements the "Subscribe" algorithm described in the GraphQL specification.
+ *
+ * Returns a Promise which resolves to either an AsyncIterator (if successful)
+ * or an ExecutionResult (error). The promise will be rejected if the schema or
+ * other arguments to this function are invalid, or if the resolved event stream
+ * is not an async iterable.
+ *
+ * If the client-provided arguments to this function do not result in a
+ * compliant subscription, a GraphQL Response (ExecutionResult) with
+ * descriptive errors and no data will be returned.
+ *
+ * If the source stream could not be created due to faulty subscription
+ * resolver logic or underlying systems, the promise will resolve to a single
+ * ExecutionResult containing `errors` and no `data`.
+ *
+ * If the operation succeeded, the promise resolves to an AsyncIterator, which
+ * yields a stream of ExecutionResults representing the response stream.
+ *
+ * Accepts either an object with named arguments, or individual arguments.
+ */
+export function subscribe(
+  args: ExecutionArgs,
+): PromiseOrValue<
+  AsyncGenerator<ExecutionResult, void, void> | ExecutionResult
+> {
+  const resultOrStream = createSourceEventStream(args);
+
+  if (isPromise(resultOrStream)) {
+    return resultOrStream.then((resolvedResultOrStream) =>
+      mapSourceToResponse(resolvedResultOrStream, args),
+    );
+  }
+
+  return mapSourceToResponse(resultOrStream, args);
+}
+
+function mapSourceToResponse(
+  resultOrStream: ExecutionResult | AsyncIterable<unknown>,
+  args: ExecutionArgs,
+): PromiseOrValue<
+  AsyncGenerator<ExecutionResult, void, void> | ExecutionResult
+> {
+  if (!isAsyncIterable(resultOrStream)) {
+    return resultOrStream;
+  }
+
+  // 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.
+  return mapAsyncIterator(resultOrStream, (payload: unknown) =>
+    execute({
+      ...args,
+      rootValue: payload,
+    }),
+  );
+}
+
+/**
+ * Implements the "CreateSourceEventStream" algorithm described in the
+ * GraphQL specification, resolving the subscription source event stream.
+ *
+ * Returns a Promise which resolves to either an AsyncIterable (if successful)
+ * or an ExecutionResult (error). The promise will be rejected if the schema or
+ * other arguments to this function are invalid, or if the resolved event stream
+ * is not an async iterable.
+ *
+ * If the client-provided arguments to this function do not result in a
+ * compliant subscription, a GraphQL Response (ExecutionResult) with
+ * descriptive errors and no data will be returned.
+ *
+ * If the the source stream could not be created due to faulty subscription
+ * resolver logic or underlying systems, the promise will resolve to a single
+ * ExecutionResult containing `errors` and no `data`.
+ *
+ * If the operation succeeded, the promise resolves to the AsyncIterable for the
+ * event stream returned by the resolver.
+ *
+ * A Source Event Stream represents a sequence of events, each of which triggers
+ * a GraphQL execution for that event.
+ *
+ * This may be useful when hosting the stateful subscription service in a
+ * different process or machine than the stateless GraphQL execution engine,
+ * or otherwise separating these two steps. For more on this, see the
+ * "Supporting Subscriptions at Scale" information in the GraphQL specification.
+ */
+export function createSourceEventStream(
+  args: ExecutionArgs,
+): PromiseOrValue<AsyncIterable<unknown> | ExecutionResult> {
+  const {
+    schema,
+    document,
+    rootValue,
+    contextValue,
+    variableValues,
+    operationName,
+    subscribeFieldResolver,
+  } = args;
+
+  // If arguments are missing or incorrectly typed, this is an internal
+  // developer mistake which should throw an early error.
+  assertValidExecutionArguments(schema, document, variableValues);
+
+  // If a valid execution context cannot be created due to incorrect arguments,
+  // a "Response" with only errors is returned.
+  const exeContext = buildExecutionContext({
+    schema,
+    document,
+    rootValue,
+    contextValue,
+    variableValues,
+    operationName,
+    subscribeFieldResolver,
+  });
+
+  // Return early errors if execution context failed.
+  if (!('schema' in exeContext)) {
+    return { errors: exeContext };
+  }
+
+  try {
+    const eventStream = executeSubscription(exeContext);
+    if (isPromise(eventStream)) {
+      return eventStream.then(undefined, (error) => ({ errors: [error] }));
+    }
+
+    return eventStream;
+  } catch (error) {
+    return { errors: [error] };
+  }
+}
+
+function executeSubscription(
+  exeContext: ExecutionContext,
+): PromiseOrValue<AsyncIterable<unknown>> {
+  const { schema, fragments, operation, variableValues, rootValue } =
+    exeContext;
+
+  const rootType = schema.getSubscriptionType();
+  if (rootType == null) {
+    throw new GraphQLError(
+      'Schema is not configured to execute subscription operation.',
+      { nodes: operation },
+    );
+  }
+
+  const rootFields = collectFields(
+    schema,
+    fragments,
+    variableValues,
+    rootType,
+    operation.selectionSet,
+  );
+  const [responseName, fieldNodes] = [...rootFields.entries()][0];
+  const fieldName = fieldNodes[0].name.value;
+  const fieldDef = schema.getField(rootType, fieldName);
+
+  if (!fieldDef) {
+    throw new GraphQLError(
+      `The subscription field "${fieldName}" is not defined.`,
+      { nodes: fieldNodes },
+    );
+  }
+
+  const path = addPath(undefined, responseName, rootType.name);
+  const info = buildResolveInfo(
+    exeContext,
+    fieldDef,
+    fieldNodes,
+    rootType,
+    path,
+  );
+
+  try {
+    // Implements the "ResolveFieldEventStream" algorithm from GraphQL specification.
+    // It differs from "ResolveFieldValue" due to providing a different `resolveFn`.
+
+    // Build a JS object of arguments from the field.arguments AST, using the
+    // variables scope to fulfill any variable references.
+    const args = getArgumentValues(fieldDef, fieldNodes[0], variableValues);
+
+    // The resolve function's optional third argument is a context value that
+    // is provided to every resolve function within an execution. It is commonly
+    // used to represent an authenticated user, or request-specific caches.
+    const contextValue = exeContext.contextValue;
+
+    // Call the `subscribe()` resolver or the default resolver to produce an
+    // AsyncIterable yielding raw payloads.
+    const resolveFn = fieldDef.subscribe ?? exeContext.subscribeFieldResolver;
+    const result = resolveFn(rootValue, args, contextValue, info);
+
+    if (isPromise(result)) {
+      return result.then(assertEventStream).then(undefined, (error) => {
+        throw locatedError(error, fieldNodes, pathToArray(path));
+      });
+    }
+
+    return assertEventStream(result);
+  } catch (error) {
+    throw locatedError(error, fieldNodes, pathToArray(path));
+  }
+}
+
+function assertEventStream(result: unknown): AsyncIterable<unknown> {
+  if (result instanceof Error) {
+    throw result;
+  }
+
+  // Assert field returned an event stream, otherwise yield an error.
+  if (!isAsyncIterable(result)) {
+    throw new GraphQLError(
+      'Subscription field must return Async Iterable. ' +
+        `Received: ${inspect(result)}.`,
+    );
+  }
+
+  return result;
+}
diff --git a/src/execution/index.ts b/src/execution/index.ts
index 7727e6d5..b27a2c29 100644
--- a/src/execution/index.ts
+++ b/src/execution/index.ts
@@ -1,10 +1,12 @@
 export { pathToArray as responsePathAsArray } from '../jsutils/Path';
 
 export {
+  createSourceEventStream,
   execute,
   executeSync,
   defaultFieldResolver,
   defaultTypeResolver,
+  subscribe,
 } from './execute';
 
 export type {
@@ -13,8 +15,6 @@ export type {
   FormattedExecutionResult,
 } from './execute';
 
-export { subscribe, createSourceEventStream } from './subscribe';
-
 export {
   getArgumentValues,
   getVariableValues,
diff --git a/src/execution/subscribe.ts b/src/execution/subscribe.ts
deleted file mode 100644
index cde4fdbd..00000000
--- a/src/execution/subscribe.ts
+++ /dev/null
@@ -1,245 +0,0 @@
-import { inspect } from '../jsutils/inspect';
-import { isAsyncIterable } from '../jsutils/isAsyncIterable';
-import { isPromise } from '../jsutils/isPromise';
-import { addPath, pathToArray } from '../jsutils/Path';
-import type { PromiseOrValue } from '../jsutils/PromiseOrValue';
-
-import { GraphQLError } from '../error/GraphQLError';
-import { locatedError } from '../error/locatedError';
-
-import { collectFields } from './collectFields';
-import type {
-  ExecutionArgs,
-  ExecutionContext,
-  ExecutionResult,
-} from './execute';
-import {
-  assertValidExecutionArguments,
-  buildExecutionContext,
-  buildResolveInfo,
-  execute,
-} from './execute';
-import { mapAsyncIterator } from './mapAsyncIterator';
-import { getArgumentValues } from './values';
-
-/**
- * Implements the "Subscribe" algorithm described in the GraphQL specification.
- *
- * Returns a Promise which resolves to either an AsyncIterator (if successful)
- * or an ExecutionResult (error). The promise will be rejected if the schema or
- * other arguments to this function are invalid, or if the resolved event stream
- * is not an async iterable.
- *
- * If the client-provided arguments to this function do not result in a
- * compliant subscription, a GraphQL Response (ExecutionResult) with
- * descriptive errors and no data will be returned.
- *
- * If the source stream could not be created due to faulty subscription
- * resolver logic or underlying systems, the promise will resolve to a single
- * ExecutionResult containing `errors` and no `data`.
- *
- * If the operation succeeded, the promise resolves to an AsyncIterator, which
- * yields a stream of ExecutionResults representing the response stream.
- *
- * Accepts either an object with named arguments, or individual arguments.
- */
-export function subscribe(
-  args: ExecutionArgs,
-): PromiseOrValue<
-  AsyncGenerator<ExecutionResult, void, void> | ExecutionResult
-> {
-  const resultOrStream = createSourceEventStream(args);
-
-  if (isPromise(resultOrStream)) {
-    return resultOrStream.then((resolvedResultOrStream) =>
-      mapSourceToResponse(resolvedResultOrStream, args),
-    );
-  }
-
-  return mapSourceToResponse(resultOrStream, args);
-}
-
-function mapSourceToResponse(
-  resultOrStream: ExecutionResult | AsyncIterable<unknown>,
-  args: ExecutionArgs,
-): PromiseOrValue<
-  AsyncGenerator<ExecutionResult, void, void> | ExecutionResult
-> {
-  if (!isAsyncIterable(resultOrStream)) {
-    return resultOrStream;
-  }
-
-  // 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.
-  return mapAsyncIterator(resultOrStream, (payload: unknown) =>
-    execute({
-      ...args,
-      rootValue: payload,
-    }),
-  );
-}
-
-/**
- * Implements the "CreateSourceEventStream" algorithm described in the
- * GraphQL specification, resolving the subscription source event stream.
- *
- * Returns a Promise which resolves to either an AsyncIterable (if successful)
- * or an ExecutionResult (error). The promise will be rejected if the schema or
- * other arguments to this function are invalid, or if the resolved event stream
- * is not an async iterable.
- *
- * If the client-provided arguments to this function do not result in a
- * compliant subscription, a GraphQL Response (ExecutionResult) with
- * descriptive errors and no data will be returned.
- *
- * If the the source stream could not be created due to faulty subscription
- * resolver logic or underlying systems, the promise will resolve to a single
- * ExecutionResult containing `errors` and no `data`.
- *
- * If the operation succeeded, the promise resolves to the AsyncIterable for the
- * event stream returned by the resolver.
- *
- * A Source Event Stream represents a sequence of events, each of which triggers
- * a GraphQL execution for that event.
- *
- * This may be useful when hosting the stateful subscription service in a
- * different process or machine than the stateless GraphQL execution engine,
- * or otherwise separating these two steps. For more on this, see the
- * "Supporting Subscriptions at Scale" information in the GraphQL specification.
- */
-export function createSourceEventStream(
-  args: ExecutionArgs,
-): PromiseOrValue<AsyncIterable<unknown> | ExecutionResult> {
-  const {
-    schema,
-    document,
-    rootValue,
-    contextValue,
-    variableValues,
-    operationName,
-    subscribeFieldResolver,
-  } = args;
-
-  // If arguments are missing or incorrectly typed, this is an internal
-  // developer mistake which should throw an early error.
-  assertValidExecutionArguments(schema, document, variableValues);
-
-  // If a valid execution context cannot be created due to incorrect arguments,
-  // a "Response" with only errors is returned.
-  const exeContext = buildExecutionContext({
-    schema,
-    document,
-    rootValue,
-    contextValue,
-    variableValues,
-    operationName,
-    subscribeFieldResolver,
-  });
-
-  // Return early errors if execution context failed.
-  if (!('schema' in exeContext)) {
-    return { errors: exeContext };
-  }
-
-  try {
-    const eventStream = executeSubscription(exeContext);
-    if (isPromise(eventStream)) {
-      return eventStream.then(undefined, (error) => ({ errors: [error] }));
-    }
-
-    return eventStream;
-  } catch (error) {
-    return { errors: [error] };
-  }
-}
-
-function executeSubscription(
-  exeContext: ExecutionContext,
-): PromiseOrValue<AsyncIterable<unknown>> {
-  const { schema, fragments, operation, variableValues, rootValue } =
-    exeContext;
-
-  const rootType = schema.getSubscriptionType();
-  if (rootType == null) {
-    throw new GraphQLError(
-      'Schema is not configured to execute subscription operation.',
-      { nodes: operation },
-    );
-  }
-
-  const rootFields = collectFields(
-    schema,
-    fragments,
-    variableValues,
-    rootType,
-    operation.selectionSet,
-  );
-  const [responseName, fieldNodes] = [...rootFields.entries()][0];
-  const fieldName = fieldNodes[0].name.value;
-  const fieldDef = schema.getField(rootType, fieldName);
-
-  if (!fieldDef) {
-    throw new GraphQLError(
-      `The subscription field "${fieldName}" is not defined.`,
-      { nodes: fieldNodes },
-    );
-  }
-
-  const path = addPath(undefined, responseName, rootType.name);
-  const info = buildResolveInfo(
-    exeContext,
-    fieldDef,
-    fieldNodes,
-    rootType,
-    path,
-  );
-
-  try {
-    // Implements the "ResolveFieldEventStream" algorithm from GraphQL specification.
-    // It differs from "ResolveFieldValue" due to providing a different `resolveFn`.
-
-    // Build a JS object of arguments from the field.arguments AST, using the
-    // variables scope to fulfill any variable references.
-    const args = getArgumentValues(fieldDef, fieldNodes[0], variableValues);
-
-    // The resolve function's optional third argument is a context value that
-    // is provided to every resolve function within an execution. It is commonly
-    // used to represent an authenticated user, or request-specific caches.
-    const contextValue = exeContext.contextValue;
-
-    // Call the `subscribe()` resolver or the default resolver to produce an
-    // AsyncIterable yielding raw payloads.
-    const resolveFn = fieldDef.subscribe ?? exeContext.subscribeFieldResolver;
-    const result = resolveFn(rootValue, args, contextValue, info);
-
-    if (isPromise(result)) {
-      return result.then(assertEventStream).then(undefined, (error) => {
-        throw locatedError(error, fieldNodes, pathToArray(path));
-      });
-    }
-
-    return assertEventStream(result);
-  } catch (error) {
-    throw locatedError(error, fieldNodes, pathToArray(path));
-  }
-}
-
-function assertEventStream(result: unknown): AsyncIterable<unknown> {
-  if (result instanceof Error) {
-    throw result;
-  }
-
-  // Assert field returned an event stream, otherwise yield an error.
-  if (!isAsyncIterable(result)) {
-    throw new GraphQLError(
-      'Subscription field must return Async Iterable. ' +
-        `Received: ${inspect(result)}.`,
-    );
-  }
-
-  return result;
-}


Test patch:
diff --git a/src/execution/__tests__/subscribe-test.ts b/src/execution/__tests__/subscribe-test.ts
index 7c18e771..6d73208c 100644
--- a/src/execution/__tests__/subscribe-test.ts
+++ b/src/execution/__tests__/subscribe-test.ts
@@ -15,7 +15,7 @@ import { GraphQLBoolean, GraphQLInt, GraphQLString } from '../../type/scalars';
 import { GraphQLSchema } from '../../type/schema';
 
 import type { ExecutionResult } from '../execute';
-import { createSourceEventStream, subscribe } from '../subscribe';
+import { createSourceEventStream, subscribe } from '../execute';
 
 import { SimplePubSub } from './simplePubSub';