Add subscriptions support to makeRemoteExectuableSchema

.vscode/settings.json

@@ -4,6 +4,7 @@
  "editor.insertSpaces": true,
  "editor.rulers": [110],
  "editor.wordWrapColumn": 110,
+ "prettier.semi": true,
  "files.trimTrailingWhitespace": true,
  "files.insertFinalNewline": true,
  "prettier.singleQuote": true,

package.json

@@ -50,13 +50,14 @@
  "dependencies": {
  "apollo-utilities": "^1.0.1",
  "deprecated-decorator": "^0.1.6",
+ "graphql-subscriptions": "^0.5.6",
  "uuid": "^3.1.0"
  },
  "peerDependencies": {
  "graphql": "^0.11.0 || ^0.12.0"
  },
  "devDependencies": {
- "@types/chai": "4.0.4",
+ "@types/chai": "4.0.10",
  "@types/graphql": "0.11.7",
  "@types/mocha": "^2.2.44",
  "@types/node": "^8.0.47",
@@ -70,13 +71,12 @@
  "graphql-subscriptions": "^0.5.4",
  "graphql-type-json": "^0.1.4",
  "istanbul": "^0.4.5",
- "iterall": "^1.1.3",
  "mocha": "^4.0.1",
  "prettier": "^1.7.4",
  "remap-istanbul": "0.9.5",
  "rimraf": "^2.6.2",
  "source-map-support": "^0.5.0",
  "tslint": "^5.8.0",
- "typescript": "2.6.1"
+ "typescript": "2.6.2"
  }
 }

src/stitching/linkToFetcher.ts

@@ -45,7 +45,7 @@ function makePromise<R>(observable: Observable<R>): Promise<R> {
  });
 }
 
-function execute(
+export function execute(
  link: ApolloLink,
  operation: GraphQLRequest,
 ): Observable<FetchResult> {

src/stitching/makeRemoteExecutableSchema.ts

@@ -1,6 +1,3 @@
-import { printSchema, Kind, ValueNode } from 'graphql';
-import linkToFetcher from './linkToFetcher';
-
 // This import doesn't actually import code - only the types.
 // Don't use ApolloLink to actually construct a link here.
 import { ApolloLink } from 'apollo-link';
@@ -20,13 +17,18 @@ import {
  ExecutionResult,
  print,
  buildSchema,
+ printSchema,
+ Kind,
+ ValueNode,
 } from 'graphql';
+import linkToFetcher, { execute } from './linkToFetcher';
 import isEmptyObject from '../isEmptyObject';
 import { IResolvers, IResolverObject } from '../Interfaces';
 import { makeExecutableSchema } from '../schemaGenerator';
 import resolveParentFromTypename from './resolveFromParentTypename';
 import defaultMergedResolver from './defaultMergedResolver';
 import { checkResultAndHandleErrors } from './errors';
+import { PubSub, ResolverFn } from 'graphql-subscriptions';
 
 export type Fetcher = (operation: FetcherOperation) => Promise<ExecutionResult>;
 
@@ -59,13 +61,16 @@ export default function makeRemoteExecutableSchema({
  typeDefs = printSchema(schema);
  }
 
+ // prepare query resolvers
+ const queryResolvers: IResolverObject = {};
  const queryType = schema.getQueryType();
  const queries = queryType.getFields();
- const queryResolvers: IResolverObject = {};
  Object.keys(queries).forEach(key => {
  queryResolvers[key] = createResolver(fetcher);
  });
- let mutationResolvers: IResolverObject = {};
+
+ // prepare mutation resolvers
+ const mutationResolvers: IResolverObject = {};
  const mutationType = schema.getMutationType();
  if (mutationType) {
  const mutations = mutationType.getFields();
@@ -74,19 +79,34 @@ export default function makeRemoteExecutableSchema({
  });
  }
 
+ // prepare subscription resolvers
+ const subscriptionResolvers: IResolverObject = {};
+ const subscriptionType = schema.getSubscriptionType();
+ if (subscriptionType) {
+ const subscriptions = subscriptionType.getFields();
+ Object.keys(subscriptions).forEach(key => {
+ subscriptionResolvers[key] = {
+ subscribe: createSubscriptionResolver(link),
+ };
+ });
+ }
+
+ // merge resolvers into resolver map
  const resolvers: IResolvers = { [queryType.name]: queryResolvers };
 
  if (!isEmptyObject(mutationResolvers)) {
  resolvers[mutationType.name] = mutationResolvers;
  }
 
+ if (!isEmptyObject(subscriptionResolvers)) {
+ resolvers[subscriptionType.name] = subscriptionResolvers;
+ }
+
+ // add missing abstract resolvers (scalar, unions, interfaces)
  const typeMap = schema.getTypeMap();
  const types = Object.keys(typeMap).map(name => typeMap[name]);
  for (const type of types) {
- if (
- type instanceof GraphQLInterfaceType ||
- type instanceof GraphQLUnionType
- ) {
+ if (type instanceof GraphQLInterfaceType || type instanceof GraphQLUnionType) {
  resolvers[type.name] = {
  __resolveType(parent, context, info) {
  return resolveParentFromTypename(parent, info.schema);
@@ -108,7 +128,8 @@ export default function makeRemoteExecutableSchema({
  type instanceof GraphQLObjectType &&
  type.name.slice(0, 2) !== '__' &&
  type !== queryType &&
- type !== mutationType
+ type !== mutationType &&
+ type !== subscriptionType
  ) {
  const resolver = {};
  Object.keys(type.getFields()).forEach(field => {
@@ -126,9 +147,7 @@ export default function makeRemoteExecutableSchema({
 
 function createResolver(fetcher: Fetcher): GraphQLFieldResolver<any, any> {
  return async (root, args, context, info) => {
- const fragments = Object.keys(info.fragments).map(
- fragment => info.fragments[fragment],
- );
+ const fragments = Object.keys(info.fragments).map(fragment => info.fragments[fragment]);
  const document = {
  kind: Kind.DOCUMENT,
  definitions: [info.operation, ...fragments],
@@ -142,6 +161,37 @@ function createResolver(fetcher: Fetcher): GraphQLFieldResolver<any, any> {
  };
 }
 
+function createSubscriptionResolver(link: ApolloLink): ResolverFn {
+ return (root, args, context, info) => {
+ const fragments = Object.keys(info.fragments).map(fragment => info.fragments[fragment]);
+ const document = {
+ kind: Kind.DOCUMENT,
+ definitions: [info.operation, ...fragments],
+ };
+
+ const operation = {
+ query: document,
+ variables: info.variableValues,
+ context: { graphqlContext: context },
+ };
+ const observable = execute(link, operation);
+
+ const pubSub = new PubSub();
+ const observer = {
+ next({ data }: any) {
+ pubSub.publish('static', data);
+ },
+ error(err: Error) {
+ pubSub.publish('static', { errors: [err] });
+ },
+ };
+
+ observable.subscribe(observer);
+
+ return pubSub.asyncIterator('static');
+ };
+}
+
 function createPassThroughScalar({
  name,
  description,