Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

move subscribe code to execute file #3636

Merged
merged 2 commits into from
Jun 12, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/execution/__tests__/subscribe-test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';

Expand Down
227 changes: 227 additions & 0 deletions src/execution/execute.ts
Original file line number Diff line number Diff line change
@@ -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';
Expand Down Expand Up @@ -51,6 +52,7 @@ import {
collectFields,
collectSubfields as _collectSubfields,
} from './collectFields';
import { mapAsyncIterator } from './mapAsyncIterator';
import { getArgumentValues, getVariableValues } from './values';

/**
Expand Down Expand Up @@ -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(
yaacovCR marked this conversation as resolved.
Show resolved Hide resolved
Expand All @@ -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(
Expand Down Expand Up @@ -543,6 +547,7 @@ function executeField(
}

/**
* TODO: consider no longer exporting this function
* @internal
*/
export function buildResolveInfo(
Expand Down Expand Up @@ -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;
}
4 changes: 2 additions & 2 deletions src/execution/index.ts
Original file line number Diff line number Diff line change
@@ -1,10 +1,12 @@
export { pathToArray as responsePathAsArray } from '../jsutils/Path';

export {
createSourceEventStream,
execute,
executeSync,
defaultFieldResolver,
defaultTypeResolver,
subscribe,
} from './execute';

export type {
Expand All @@ -13,8 +15,6 @@ export type {
FormattedExecutionResult,
} from './execute';

export { subscribe, createSourceEventStream } from './subscribe';

export {
getArgumentValues,
getVariableValues,
Expand Down
Loading