Project files

This commit is contained in:
2023-11-09 18:47:11 +01:00
parent 695abe054b
commit c415135aae
8554 changed files with 858111 additions and 0 deletions

View File

@@ -0,0 +1,5 @@
export {
subscribe,
createSourceEventStream,
SubscriptionArgs,
} from './subscribe';

View File

@@ -0,0 +1,19 @@
"use strict";
Object.defineProperty(exports, "__esModule", {
value: true
});
Object.defineProperty(exports, "subscribe", {
enumerable: true,
get: function get() {
return _subscribe.subscribe;
}
});
Object.defineProperty(exports, "createSourceEventStream", {
enumerable: true,
get: function get() {
return _subscribe.createSourceEventStream;
}
});
var _subscribe = require("./subscribe.js");

View File

@@ -0,0 +1,3 @@
// @flow strict
export { subscribe, createSourceEventStream } from './subscribe';
export type { SubscriptionArgs } from './subscribe';

View File

@@ -0,0 +1 @@
export { subscribe, createSourceEventStream } from "./subscribe.mjs";

View File

@@ -0,0 +1,11 @@
import { PromiseOrValue } from '../jsutils/PromiseOrValue';
/**
* Given an AsyncIterable and a callback function, return an AsyncIterator
* which produces values mapped via calling the callback function.
*/
export default function mapAsyncIterator<T, U>(
iterable: AsyncIterable<T>,
callback: (arg: T) => PromiseOrValue<U>,
rejectCallback?: (arg: any) => PromiseOrValue<U>,
): any; // TS_SPECIFIC: AsyncGenerator requires typescript@3.6

View File

@@ -0,0 +1,86 @@
"use strict";
Object.defineProperty(exports, "__esModule", {
value: true
});
exports.default = mapAsyncIterator;
var _symbols = require("../polyfills/symbols.js");
function _defineProperty(obj, key, value) { if (key in obj) { Object.defineProperty(obj, key, { value: value, enumerable: true, configurable: true, writable: true }); } else { obj[key] = value; } return obj; }
/**
* Given an AsyncIterable and a callback function, return an AsyncIterator
* which produces values mapped via calling the callback function.
*/
function mapAsyncIterator(iterable, callback, rejectCallback) {
// $FlowFixMe[prop-missing]
var iteratorMethod = iterable[_symbols.SYMBOL_ASYNC_ITERATOR];
var iterator = iteratorMethod.call(iterable);
var $return;
var abruptClose;
if (typeof iterator.return === 'function') {
$return = iterator.return;
abruptClose = function abruptClose(error) {
var rethrow = function rethrow() {
return Promise.reject(error);
};
return $return.call(iterator).then(rethrow, rethrow);
};
}
function mapResult(result) {
return result.done ? result : asyncMapValue(result.value, callback).then(iteratorResult, abruptClose);
}
var mapReject;
if (rejectCallback) {
// Capture rejectCallback to ensure it cannot be null.
var reject = rejectCallback;
mapReject = function mapReject(error) {
return asyncMapValue(error, reject).then(iteratorResult, abruptClose);
};
}
/* TODO: Flow doesn't support symbols as keys:
https://github.com/facebook/flow/issues/3258 */
return _defineProperty({
next: function next() {
return iterator.next().then(mapResult, mapReject);
},
return: function _return() {
return $return ? $return.call(iterator).then(mapResult, mapReject) : Promise.resolve({
value: undefined,
done: true
});
},
throw: function _throw(error) {
if (typeof iterator.throw === 'function') {
return iterator.throw(error).then(mapResult, mapReject);
}
return Promise.reject(error).catch(abruptClose);
}
}, _symbols.SYMBOL_ASYNC_ITERATOR, function () {
return this;
});
}
function asyncMapValue(value, callback) {
return new Promise(function (resolve) {
return resolve(callback(value));
});
}
function iteratorResult(value) {
return {
value: value,
done: false
};
}

View File

@@ -0,0 +1,74 @@
// @flow strict
import { SYMBOL_ASYNC_ITERATOR } from '../polyfills/symbols';
import type { PromiseOrValue } from '../jsutils/PromiseOrValue';
/**
* Given an AsyncIterable and a callback function, return an AsyncIterator
* which produces values mapped via calling the callback function.
*/
export default function mapAsyncIterator<T, U>(
iterable: AsyncIterable<T> | AsyncGenerator<T, void, void>,
callback: (T) => PromiseOrValue<U>,
rejectCallback?: (any) => PromiseOrValue<U>,
): AsyncGenerator<U, void, void> {
// $FlowFixMe[prop-missing]
const iteratorMethod = iterable[SYMBOL_ASYNC_ITERATOR];
const iterator: any = iteratorMethod.call(iterable);
let $return: any;
let abruptClose;
if (typeof iterator.return === 'function') {
$return = iterator.return;
abruptClose = (error: mixed) => {
const rethrow = () => Promise.reject(error);
return $return.call(iterator).then(rethrow, rethrow);
};
}
function mapResult(result: IteratorResult<T, void>) {
return result.done
? result
: asyncMapValue(result.value, callback).then(iteratorResult, abruptClose);
}
let mapReject;
if (rejectCallback) {
// Capture rejectCallback to ensure it cannot be null.
const reject = rejectCallback;
mapReject = (error: mixed) =>
asyncMapValue(error, reject).then(iteratorResult, abruptClose);
}
/* TODO: Flow doesn't support symbols as keys:
https://github.com/facebook/flow/issues/3258 */
return ({
next(): Promise<IteratorResult<U, void>> {
return iterator.next().then(mapResult, mapReject);
},
return() {
return $return
? $return.call(iterator).then(mapResult, mapReject)
: Promise.resolve({ value: undefined, done: true });
},
throw(error?: mixed): Promise<IteratorResult<U, void>> {
if (typeof iterator.throw === 'function') {
return iterator.throw(error).then(mapResult, mapReject);
}
return Promise.reject(error).catch(abruptClose);
},
[SYMBOL_ASYNC_ITERATOR]() {
return this;
},
}: $FlowFixMe);
}
function asyncMapValue<T, U>(
value: T,
callback: (T) => PromiseOrValue<U>,
): Promise<U> {
return new Promise((resolve) => resolve(callback(value)));
}
function iteratorResult<T>(value: T): IteratorResult<T, void> {
return { value, done: false };
}

View File

@@ -0,0 +1,79 @@
function _defineProperty(obj, key, value) { if (key in obj) { Object.defineProperty(obj, key, { value: value, enumerable: true, configurable: true, writable: true }); } else { obj[key] = value; } return obj; }
import { SYMBOL_ASYNC_ITERATOR } from "../polyfills/symbols.mjs";
/**
* Given an AsyncIterable and a callback function, return an AsyncIterator
* which produces values mapped via calling the callback function.
*/
export default function mapAsyncIterator(iterable, callback, rejectCallback) {
// $FlowFixMe[prop-missing]
var iteratorMethod = iterable[SYMBOL_ASYNC_ITERATOR];
var iterator = iteratorMethod.call(iterable);
var $return;
var abruptClose;
if (typeof iterator.return === 'function') {
$return = iterator.return;
abruptClose = function abruptClose(error) {
var rethrow = function rethrow() {
return Promise.reject(error);
};
return $return.call(iterator).then(rethrow, rethrow);
};
}
function mapResult(result) {
return result.done ? result : asyncMapValue(result.value, callback).then(iteratorResult, abruptClose);
}
var mapReject;
if (rejectCallback) {
// Capture rejectCallback to ensure it cannot be null.
var reject = rejectCallback;
mapReject = function mapReject(error) {
return asyncMapValue(error, reject).then(iteratorResult, abruptClose);
};
}
/* TODO: Flow doesn't support symbols as keys:
https://github.com/facebook/flow/issues/3258 */
return _defineProperty({
next: function next() {
return iterator.next().then(mapResult, mapReject);
},
return: function _return() {
return $return ? $return.call(iterator).then(mapResult, mapReject) : Promise.resolve({
value: undefined,
done: true
});
},
throw: function _throw(error) {
if (typeof iterator.throw === 'function') {
return iterator.throw(error).then(mapResult, mapReject);
}
return Promise.reject(error).catch(abruptClose);
}
}, SYMBOL_ASYNC_ITERATOR, function () {
return this;
});
}
function asyncMapValue(value, callback) {
return new Promise(function (resolve) {
return resolve(callback(value));
});
}
function iteratorResult(value) {
return {
value: value,
done: false
};
}

View File

@@ -0,0 +1,80 @@
import { Maybe } from '../jsutils/Maybe';
import { DocumentNode } from '../language/ast';
import { ExecutionResult } from '../execution/execute';
import { GraphQLSchema } from '../type/schema';
import { GraphQLFieldResolver } from '../type/definition';
export interface SubscriptionArgs {
schema: GraphQLSchema;
document: DocumentNode;
rootValue?: any;
contextValue?: any;
variableValues?: Maybe<Record<string, any>>;
operationName?: Maybe<string>;
fieldResolver?: Maybe<GraphQLFieldResolver<any, any>>;
subscribeFieldResolver?: Maybe<GraphQLFieldResolver<any, any>>;
}
/**
* Implements the "Subscribe" algorithm described in the GraphQL specification.
*
* Returns a Promise which resolves to either an AsyncIterator (if successful)
* or an ExecutionResult (client error). The promise will be rejected if a
* server error occurs.
*
* 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 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: SubscriptionArgs,
): Promise<AsyncIterableIterator<ExecutionResult> | ExecutionResult>;
export function subscribe(
schema: GraphQLSchema,
document: DocumentNode,
rootValue?: any,
contextValue?: any,
variableValues?: Maybe<{ [key: string]: any }>,
operationName?: Maybe<string>,
fieldResolver?: Maybe<GraphQLFieldResolver<any, any>>,
subscribeFieldResolver?: Maybe<GraphQLFieldResolver<any, any>>,
): Promise<AsyncIterableIterator<ExecutionResult> | ExecutionResult>;
/**
* Implements the "CreateSourceEventStream" algorithm described in the
* GraphQL specification, resolving the subscription source event stream.
*
* Returns a Promise<AsyncIterable>.
*
* If the client-provided invalid arguments, the source stream could not be
* created, or the resolver did not return an AsyncIterable, this function will
* will throw an error, which should be caught and handled by the caller.
*
* 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(
schema: GraphQLSchema,
document: DocumentNode,
rootValue?: any,
contextValue?: any,
variableValues?: { [key: string]: any },
operationName?: Maybe<string>,
fieldResolver?: Maybe<GraphQLFieldResolver<any, any>>,
): Promise<AsyncIterable<any> | ExecutionResult>;

View File

@@ -0,0 +1,192 @@
"use strict";
Object.defineProperty(exports, "__esModule", {
value: true
});
exports.subscribe = subscribe;
exports.createSourceEventStream = createSourceEventStream;
var _inspect = _interopRequireDefault(require("../jsutils/inspect.js"));
var _isAsyncIterable = _interopRequireDefault(require("../jsutils/isAsyncIterable.js"));
var _Path = require("../jsutils/Path.js");
var _GraphQLError = require("../error/GraphQLError.js");
var _locatedError = require("../error/locatedError.js");
var _values = require("../execution/values.js");
var _execute = require("../execution/execute.js");
var _getOperationRootType = require("../utilities/getOperationRootType.js");
var _mapAsyncIterator = _interopRequireDefault(require("./mapAsyncIterator.js"));
function _interopRequireDefault(obj) { return obj && obj.__esModule ? obj : { default: obj }; }
function subscribe(argsOrSchema, document, rootValue, contextValue, variableValues, operationName, fieldResolver, subscribeFieldResolver) {
/* eslint-enable no-redeclare */
// Extract arguments from object args if provided.
return arguments.length === 1 ? subscribeImpl(argsOrSchema) : subscribeImpl({
schema: argsOrSchema,
document: document,
rootValue: rootValue,
contextValue: contextValue,
variableValues: variableValues,
operationName: operationName,
fieldResolver: fieldResolver,
subscribeFieldResolver: subscribeFieldResolver
});
}
/**
* This function checks if the error is a GraphQLError. If it is, report it as
* an ExecutionResult, containing only errors and no data. Otherwise treat the
* error as a system-class error and re-throw it.
*/
function reportGraphQLError(error) {
if (error instanceof _GraphQLError.GraphQLError) {
return {
errors: [error]
};
}
throw error;
}
function subscribeImpl(args) {
var schema = args.schema,
document = args.document,
rootValue = args.rootValue,
contextValue = args.contextValue,
variableValues = args.variableValues,
operationName = args.operationName,
fieldResolver = args.fieldResolver,
subscribeFieldResolver = args.subscribeFieldResolver;
var sourcePromise = createSourceEventStream(schema, document, rootValue, contextValue, variableValues, operationName, subscribeFieldResolver); // 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.
var mapSourceToResponse = function mapSourceToResponse(payload) {
return (0, _execute.execute)({
schema: schema,
document: document,
rootValue: payload,
contextValue: contextValue,
variableValues: variableValues,
operationName: operationName,
fieldResolver: fieldResolver
});
}; // Resolve the Source Stream, then map every source value to a
// ExecutionResult value as described above.
return sourcePromise.then(function (resultOrStream) {
return (// Note: Flow can't refine isAsyncIterable, so explicit casts are used.
(0, _isAsyncIterable.default)(resultOrStream) ? (0, _mapAsyncIterator.default)(resultOrStream, mapSourceToResponse, reportGraphQLError) : resultOrStream
);
});
}
/**
* 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.
*/
function createSourceEventStream(schema, document, rootValue, contextValue, variableValues, operationName, fieldResolver) {
// If arguments are missing or incorrectly typed, this is an internal
// developer mistake which should throw an early error.
(0, _execute.assertValidExecutionArguments)(schema, document, variableValues);
return new Promise(function (resolve) {
// If a valid context cannot be created due to incorrect arguments,
// this will throw an error.
var exeContext = (0, _execute.buildExecutionContext)(schema, document, rootValue, contextValue, variableValues, operationName, fieldResolver);
resolve( // Return early errors if execution context failed.
Array.isArray(exeContext) ? {
errors: exeContext
} : executeSubscription(exeContext));
}).catch(reportGraphQLError);
}
function executeSubscription(exeContext) {
var schema = exeContext.schema,
operation = exeContext.operation,
variableValues = exeContext.variableValues,
rootValue = exeContext.rootValue;
var type = (0, _getOperationRootType.getOperationRootType)(schema, operation);
var fields = (0, _execute.collectFields)(exeContext, type, operation.selectionSet, Object.create(null), Object.create(null));
var responseNames = Object.keys(fields);
var responseName = responseNames[0];
var fieldNodes = fields[responseName];
var fieldNode = fieldNodes[0];
var fieldName = fieldNode.name.value;
var fieldDef = (0, _execute.getFieldDef)(schema, type, fieldName);
if (!fieldDef) {
throw new _GraphQLError.GraphQLError("The subscription field \"".concat(fieldName, "\" is not defined."), fieldNodes);
}
var path = (0, _Path.addPath)(undefined, responseName, type.name);
var info = (0, _execute.buildResolveInfo)(exeContext, fieldDef, fieldNodes, type, path); // Coerce to Promise for easier error handling and consistent return type.
return new Promise(function (resolveResult) {
var _fieldDef$subscribe;
// 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.
var args = (0, _values.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.
var contextValue = exeContext.contextValue; // Call the `subscribe()` resolver or the default resolver to produce an
// AsyncIterable yielding raw payloads.
var resolveFn = (_fieldDef$subscribe = fieldDef.subscribe) !== null && _fieldDef$subscribe !== void 0 ? _fieldDef$subscribe : exeContext.fieldResolver;
resolveResult(resolveFn(rootValue, args, contextValue, info));
}).then(function (eventStream) {
if (eventStream instanceof Error) {
throw (0, _locatedError.locatedError)(eventStream, fieldNodes, (0, _Path.pathToArray)(path));
} // Assert field returned an event stream, otherwise yield an error.
if (!(0, _isAsyncIterable.default)(eventStream)) {
throw new Error('Subscription field must return Async Iterable. ' + "Received: ".concat((0, _inspect.default)(eventStream), "."));
}
return eventStream;
}, function (error) {
throw (0, _locatedError.locatedError)(error, fieldNodes, (0, _Path.pathToArray)(path));
});
}

View File

@@ -0,0 +1,298 @@
// @flow strict
import inspect from '../jsutils/inspect';
import isAsyncIterable from '../jsutils/isAsyncIterable';
import { addPath, pathToArray } from '../jsutils/Path';
import { GraphQLError } from '../error/GraphQLError';
import { locatedError } from '../error/locatedError';
import type { DocumentNode } from '../language/ast';
import type { ExecutionResult, ExecutionContext } from '../execution/execute';
import { getArgumentValues } from '../execution/values';
import {
assertValidExecutionArguments,
buildExecutionContext,
buildResolveInfo,
collectFields,
execute,
getFieldDef,
} from '../execution/execute';
import type { GraphQLSchema } from '../type/schema';
import type { GraphQLFieldResolver } from '../type/definition';
import { getOperationRootType } from '../utilities/getOperationRootType';
import mapAsyncIterator from './mapAsyncIterator';
export type SubscriptionArgs = {|
schema: GraphQLSchema,
document: DocumentNode,
rootValue?: mixed,
contextValue?: mixed,
variableValues?: ?{ +[variable: string]: mixed, ... },
operationName?: ?string,
fieldResolver?: ?GraphQLFieldResolver<any, any>,
subscribeFieldResolver?: ?GraphQLFieldResolver<any, any>,
|};
/**
* 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.
*/
declare function subscribe(
SubscriptionArgs,
..._: []
): Promise<AsyncGenerator<ExecutionResult, void, void> | ExecutionResult>;
/* eslint-disable no-redeclare */
declare function subscribe(
schema: GraphQLSchema,
document: DocumentNode,
rootValue?: mixed,
contextValue?: mixed,
variableValues?: ?{ +[variable: string]: mixed, ... },
operationName?: ?string,
fieldResolver?: ?GraphQLFieldResolver<any, any>,
subscribeFieldResolver?: ?GraphQLFieldResolver<any, any>,
): Promise<AsyncIterator<ExecutionResult> | ExecutionResult>;
export function subscribe(
argsOrSchema,
document,
rootValue,
contextValue,
variableValues,
operationName,
fieldResolver,
subscribeFieldResolver,
) {
/* eslint-enable no-redeclare */
// Extract arguments from object args if provided.
return arguments.length === 1
? subscribeImpl(argsOrSchema)
: subscribeImpl({
schema: argsOrSchema,
document,
rootValue,
contextValue,
variableValues,
operationName,
fieldResolver,
subscribeFieldResolver,
});
}
/**
* This function checks if the error is a GraphQLError. If it is, report it as
* an ExecutionResult, containing only errors and no data. Otherwise treat the
* error as a system-class error and re-throw it.
*/
function reportGraphQLError(error: mixed): ExecutionResult {
if (error instanceof GraphQLError) {
return { errors: [error] };
}
throw error;
}
function subscribeImpl(
args: SubscriptionArgs,
): Promise<AsyncGenerator<ExecutionResult, void, void> | ExecutionResult> {
const {
schema,
document,
rootValue,
contextValue,
variableValues,
operationName,
fieldResolver,
subscribeFieldResolver,
} = args;
const sourcePromise = createSourceEventStream(
schema,
document,
rootValue,
contextValue,
variableValues,
operationName,
subscribeFieldResolver,
);
// 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.
const mapSourceToResponse = (payload) =>
execute({
schema,
document,
rootValue: payload,
contextValue,
variableValues,
operationName,
fieldResolver,
});
// Resolve the Source Stream, then map every source value to a
// ExecutionResult value as described above.
return sourcePromise.then((resultOrStream) =>
// Note: Flow can't refine isAsyncIterable, so explicit casts are used.
isAsyncIterable(resultOrStream)
? mapAsyncIterator(
resultOrStream,
mapSourceToResponse,
reportGraphQLError,
)
: ((resultOrStream: any): ExecutionResult),
);
}
/**
* 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(
schema: GraphQLSchema,
document: DocumentNode,
rootValue?: mixed,
contextValue?: mixed,
variableValues?: ?{ +[variable: string]: mixed, ... },
operationName?: ?string,
fieldResolver?: ?GraphQLFieldResolver<any, any>,
): Promise<AsyncIterable<mixed> | ExecutionResult> {
// If arguments are missing or incorrectly typed, this is an internal
// developer mistake which should throw an early error.
assertValidExecutionArguments(schema, document, variableValues);
return new Promise((resolve) => {
// If a valid context cannot be created due to incorrect arguments,
// this will throw an error.
const exeContext = buildExecutionContext(
schema,
document,
rootValue,
contextValue,
variableValues,
operationName,
fieldResolver,
);
resolve(
// Return early errors if execution context failed.
Array.isArray(exeContext)
? { errors: exeContext }
: executeSubscription(exeContext),
);
}).catch(reportGraphQLError);
}
function executeSubscription(
exeContext: ExecutionContext,
): Promise<AsyncIterable<mixed>> {
const { schema, operation, variableValues, rootValue } = exeContext;
const type = getOperationRootType(schema, operation);
const fields = collectFields(
exeContext,
type,
operation.selectionSet,
Object.create(null),
Object.create(null),
);
const responseNames = Object.keys(fields);
const responseName = responseNames[0];
const fieldNodes = fields[responseName];
const fieldNode = fieldNodes[0];
const fieldName = fieldNode.name.value;
const fieldDef = getFieldDef(schema, type, fieldName);
if (!fieldDef) {
throw new GraphQLError(
`The subscription field "${fieldName}" is not defined.`,
fieldNodes,
);
}
const path = addPath(undefined, responseName, type.name);
const info = buildResolveInfo(exeContext, fieldDef, fieldNodes, type, path);
// Coerce to Promise for easier error handling and consistent return type.
return new Promise((resolveResult) => {
// 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.fieldResolver;
resolveResult(resolveFn(rootValue, args, contextValue, info));
}).then(
(eventStream) => {
if (eventStream instanceof Error) {
throw locatedError(eventStream, fieldNodes, pathToArray(path));
}
// Assert field returned an event stream, otherwise yield an error.
if (!isAsyncIterable(eventStream)) {
throw new Error(
'Subscription field must return Async Iterable. ' +
`Received: ${inspect(eventStream)}.`,
);
}
return eventStream;
},
(error) => {
throw locatedError(error, fieldNodes, pathToArray(path));
},
);
}

View File

@@ -0,0 +1,172 @@
import inspect from "../jsutils/inspect.mjs";
import isAsyncIterable from "../jsutils/isAsyncIterable.mjs";
import { addPath, pathToArray } from "../jsutils/Path.mjs";
import { GraphQLError } from "../error/GraphQLError.mjs";
import { locatedError } from "../error/locatedError.mjs";
import { getArgumentValues } from "../execution/values.mjs";
import { assertValidExecutionArguments, buildExecutionContext, buildResolveInfo, collectFields, execute, getFieldDef } from "../execution/execute.mjs";
import { getOperationRootType } from "../utilities/getOperationRootType.mjs";
import mapAsyncIterator from "./mapAsyncIterator.mjs";
export function subscribe(argsOrSchema, document, rootValue, contextValue, variableValues, operationName, fieldResolver, subscribeFieldResolver) {
/* eslint-enable no-redeclare */
// Extract arguments from object args if provided.
return arguments.length === 1 ? subscribeImpl(argsOrSchema) : subscribeImpl({
schema: argsOrSchema,
document: document,
rootValue: rootValue,
contextValue: contextValue,
variableValues: variableValues,
operationName: operationName,
fieldResolver: fieldResolver,
subscribeFieldResolver: subscribeFieldResolver
});
}
/**
* This function checks if the error is a GraphQLError. If it is, report it as
* an ExecutionResult, containing only errors and no data. Otherwise treat the
* error as a system-class error and re-throw it.
*/
function reportGraphQLError(error) {
if (error instanceof GraphQLError) {
return {
errors: [error]
};
}
throw error;
}
function subscribeImpl(args) {
var schema = args.schema,
document = args.document,
rootValue = args.rootValue,
contextValue = args.contextValue,
variableValues = args.variableValues,
operationName = args.operationName,
fieldResolver = args.fieldResolver,
subscribeFieldResolver = args.subscribeFieldResolver;
var sourcePromise = createSourceEventStream(schema, document, rootValue, contextValue, variableValues, operationName, subscribeFieldResolver); // 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.
var mapSourceToResponse = function mapSourceToResponse(payload) {
return execute({
schema: schema,
document: document,
rootValue: payload,
contextValue: contextValue,
variableValues: variableValues,
operationName: operationName,
fieldResolver: fieldResolver
});
}; // Resolve the Source Stream, then map every source value to a
// ExecutionResult value as described above.
return sourcePromise.then(function (resultOrStream) {
return (// Note: Flow can't refine isAsyncIterable, so explicit casts are used.
isAsyncIterable(resultOrStream) ? mapAsyncIterator(resultOrStream, mapSourceToResponse, reportGraphQLError) : resultOrStream
);
});
}
/**
* 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(schema, document, rootValue, contextValue, variableValues, operationName, fieldResolver) {
// If arguments are missing or incorrectly typed, this is an internal
// developer mistake which should throw an early error.
assertValidExecutionArguments(schema, document, variableValues);
return new Promise(function (resolve) {
// If a valid context cannot be created due to incorrect arguments,
// this will throw an error.
var exeContext = buildExecutionContext(schema, document, rootValue, contextValue, variableValues, operationName, fieldResolver);
resolve( // Return early errors if execution context failed.
Array.isArray(exeContext) ? {
errors: exeContext
} : executeSubscription(exeContext));
}).catch(reportGraphQLError);
}
function executeSubscription(exeContext) {
var schema = exeContext.schema,
operation = exeContext.operation,
variableValues = exeContext.variableValues,
rootValue = exeContext.rootValue;
var type = getOperationRootType(schema, operation);
var fields = collectFields(exeContext, type, operation.selectionSet, Object.create(null), Object.create(null));
var responseNames = Object.keys(fields);
var responseName = responseNames[0];
var fieldNodes = fields[responseName];
var fieldNode = fieldNodes[0];
var fieldName = fieldNode.name.value;
var fieldDef = getFieldDef(schema, type, fieldName);
if (!fieldDef) {
throw new GraphQLError("The subscription field \"".concat(fieldName, "\" is not defined."), fieldNodes);
}
var path = addPath(undefined, responseName, type.name);
var info = buildResolveInfo(exeContext, fieldDef, fieldNodes, type, path); // Coerce to Promise for easier error handling and consistent return type.
return new Promise(function (resolveResult) {
var _fieldDef$subscribe;
// 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.
var 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.
var contextValue = exeContext.contextValue; // Call the `subscribe()` resolver or the default resolver to produce an
// AsyncIterable yielding raw payloads.
var resolveFn = (_fieldDef$subscribe = fieldDef.subscribe) !== null && _fieldDef$subscribe !== void 0 ? _fieldDef$subscribe : exeContext.fieldResolver;
resolveResult(resolveFn(rootValue, args, contextValue, info));
}).then(function (eventStream) {
if (eventStream instanceof Error) {
throw locatedError(eventStream, fieldNodes, pathToArray(path));
} // Assert field returned an event stream, otherwise yield an error.
if (!isAsyncIterable(eventStream)) {
throw new Error('Subscription field must return Async Iterable. ' + "Received: ".concat(inspect(eventStream), "."));
}
return eventStream;
}, function (error) {
throw locatedError(error, fieldNodes, pathToArray(path));
});
}