GraphQL websocket subscription integration

CHANGELOG.md

@@ -3,15 +3,19 @@
 Expect active development and potentially significant breaking changes in the `0.x` track. We'll try to be diligent about releasing a `1.0` version in a timely fashion (ideally within 3 to 6 months), to signal the start of a more stable API.
 
 ### vNEXT
+
+
+### v0.4.12
 - Fixed an issue with named fragments in batched queries. [PR #509](https://github.com/apollostack/apollo-client/pull/509) and [Issue #501](https://github.com/apollostack/apollo-client/issues/501).
 - Fixed an issue with unused variables in queries after diffing queries against information available in the store. [PR #518](https://github.com/apollostack/apollo-client/pull/518) and [Issue #496](https://github.com/apollostack/apollo-client/issues/496).
+- Add code to support GraphQL subscriptions. [PR #540](https://github.com/apollostack/apollo-client/pull/540).
 - Fixed a couple of issues within query merging that caused issues with null values or arrays in responses. [PR #523](https://github.com/apollostack/apollo-client/pull/523).
+- Added an `updateQuery` method on observable queries. Allows application code to arbitrary change the result of a query normalized to store, without issuing any network requests. [PR #506](https://github.com/apollostack/apollo-client/pull/506) and [Issue #495](https://github.com/apollostack/apollo-client/issues/495).
 
 ### v0.4.11
 
 - Added an `refetchQueries` option to `mutate`. The point is to just refetch certain queries on a mutation rather than having to manually specify how the result should be incorporated for each of them with `updateQueries`. [PR #482](https://github.com/apollostack/apollo-client/pull/482) and [Issue #448](https://github.com/apollostack/apollo-client/issues/448).
 - Print errors produced by application-supplied reducer functions passed to `updateQueries` or `updateQuery` options for `mutate` or `fetchMore` respectively. [PR #500](https://github.com/apollostack/apollo-client/pull/500) and [Issue #479](https://github.com/apollostack/apollo-client/issues/479).
-- Added an `updateQuery` method on observable queries. Allows application code to arbitrary change the result of a query normalized to store, without issuing any network requests. [PR #506](https://github.com/apollostack/apollo-client/pull/506) and [Issue #495](https://github.com/apollostack/apollo-client/issues/495).
 
 ### v0.4.10
 

src/ObservableQuery.ts

@@ -1,4 +1,4 @@
-import { WatchQueryOptions, FetchMoreQueryOptions } from './watchQueryOptions';
+import { WatchQueryOptions, FetchMoreQueryOptions, GraphQLSubscriptionOptions } from './watchQueryOptions';
 
 import { Observable, Observer } from './util/Observable';
 
@@ -32,6 +32,7 @@ export interface UpdateQueryOptions {
 export class ObservableQuery extends Observable<ApolloQueryResult> {
   public refetch: (variables?: any) => Promise<ApolloQueryResult>;
   public fetchMore: (options: FetchMoreQueryOptions & FetchMoreOptions) => Promise<any>;
+  public startGraphQLSubscription: (options: GraphQLSubscriptionOptions) => number;
   public updateQuery: (mapFn: (previousQueryResult: any, options: UpdateQueryOptions) => any) => void;
   public stopPolling: () => void;
   public startPolling: (p: number) => void;
@@ -144,14 +145,45 @@ export class ObservableQuery extends Observable<ApolloQueryResult> {
         });
     };
 
+    this.startGraphQLSubscription = (graphQLSubscriptionOptions: GraphQLSubscriptionOptions) => {
+
+      const subOptions = {
+        query: graphQLSubscriptionOptions.subscription,
+        // TODO: test variables and fragments?
+        variables: graphQLSubscriptionOptions.variables,
+        fragments: graphQLSubscriptionOptions.fragments,
+        handler: (error: Object, result: Object) => {
+          const reducer = graphQLSubscriptionOptions.updateFunction;
+          if (error) {
+            throw new Error(JSON.stringify(error));
+          } else {
+             const mapFn = (previousResult, { queryVariables }) => {
+              return reducer(
+                previousResult, {
+                  subscriptionResult: result,
+                  queryVariables,
+                }
+              );
+            };
+            this.updateQuery(mapFn);
+          }
+
+        },
+      };
+
+      if (graphQLSubscriptionOptions) {
+        return this.queryManager.startSubscription(subOptions);
+      };
+      return null;
+    };
+
     this.updateQuery = (mapFn) => {
       const {
         previousResult,
         queryVariables,
         querySelectionSet,
         queryFragments = [],
       } = this.queryManager.getQueryWithPreviousResult(this.queryId);
-
       const newResult = tryFunctionOrLogError(
         () => mapFn(previousResult, { queryVariables }));
 

src/QueryManager.ts

@@ -1,5 +1,6 @@
 import {
   NetworkInterface,
+  SubscriptionNetworkInterface,
   Request,
 } from './networkInterface';
 
@@ -95,6 +96,13 @@ import { ObservableQuery } from './ObservableQuery';
 
 export type QueryListener = (queryStoreValue: QueryStoreValue) => void;
 
+ export interface SubscriptionOptions {
+  query: Document;
+  variables?: { [key: string]: any };
+  fragments?: FragmentDefinition[];
+  handler: (error: Object, result: Object) => void;
+};
+
 export class QueryManager {
   public pollingTimers: {[queryId: string]: NodeJS.Timer | any}; //oddity in Typescript
   public scheduler: QueryScheduler;
@@ -523,6 +531,32 @@ export class QueryManager {
     return queryId;
   }
 
+  public startSubscription(
+    options: SubscriptionOptions
+  ): number {
+    const {
+      query,
+      variables,
+      fragments = [],
+      handler,
+    } = options;
+
+    let queryDoc = addFragmentsToDocument(query, fragments);
+    // Apply the query transformer if one has been provided.
+    if (this.queryTransformer) {
+      queryDoc = applyTransformers(queryDoc, [this.queryTransformer]);
+    }
+    const request: Request = {
+      query: queryDoc,
+      variables,
+      operationName: getOperationName(queryDoc),
+    };
+
+    // QueryManager sets up the handler so the query can be transformed. Alternatively,
+    // pass in the transformer to the ObservableQuery.
+    return (this.networkInterface as SubscriptionNetworkInterface).subscribe(request, handler);
+  };
+
   public stopQuery(queryId: string) {
     // XXX in the future if we should cancel the request
     // so that it never tries to return data

src/networkInterface.ts

@@ -38,6 +38,11 @@ export interface NetworkInterface {
   query(request: Request): Promise<GraphQLResult>;
 }
 
+export interface SubscriptionNetworkInterface extends NetworkInterface {
+  subscribe(request: Request, handler: (error, result) => void): number;
+  unsubscribe(id: Number): void;
+}
+
 export interface BatchedNetworkInterface extends NetworkInterface {
   batchQuery(requests: Request[]): Promise<GraphQLResult[]>;
 }

src/watchQueryOptions.ts

@@ -17,3 +17,13 @@ export interface FetchMoreQueryOptions {
   query?: Document;
   variables?: { [key: string]: any };
 }
+
+export interface GraphQLSubscriptionOptions {
+  subscription: Document;
+  variables?: { [key: string]: any };
+  fragments?: FragmentDefinition[];
+  updateFunction: (previousQueryResult: Object, options: {
+    subscriptionResult: Object,
+    queryVariables: Object,
+  }) => Object;
+};

test/graphqlSubscriptions.ts

@@ -0,0 +1,233 @@
+import {
+  mockSubscriptionNetworkInterface,
+} from './mocks/mockNetworkInterface';
+
+import {
+  assert,
+} from 'chai';
+
+import clonedeep = require('lodash.clonedeep');
+
+import ApolloClient from '../src';
+
+import gql from 'graphql-tag';
+
+import {
+  QueryManager,
+} from '../src/QueryManager';
+
+import {
+  createApolloStore,
+} from '../src/store';
+
+describe('GraphQL Subscriptions', () => {
+  const results = ['Dahivat Pandya', 'Vyacheslav Kim', 'Changping Chen', 'Amanda Liu'].map(
+    name => ({ result: { user: { name: name } }, delay: 10 })
+  );
+
+  let sub1;
+  let options;
+  let watchQueryOptions;
+  let sub2;
+  let commentsQuery;
+  let commentsVariables;
+  let commentsSub;
+  let commentsResult;
+  let commentsResultMore;
+  let commentsWatchQueryOptions;
+  beforeEach(() => {
+
+    sub1 = {
+      request: {
+        query: gql`
+          subscription UserInfo($name: String) {
+            user(name: $name) {
+              name
+            }
+          }
+        `,
+        variables: {
+          name: 'Changping Chen',
+        },
+      },
+      id: 0,
+      results: [...results],
+    };
+
+    options = {
+      query: gql`
+        subscription UserInfo($name: String) {
+          user(name: $name) {
+            name
+          }
+        }
+      `,
+      variables: {
+          name: 'Changping Chen',
+        },
+      handler: (error, result) => {
+        // do nothing
+      },
+    };
+
+    watchQueryOptions = {
+      query: gql`
+        query UserInfo($name: String) {
+          user(name: $name) {
+            name
+          }
+        }
+      `,
+      variables: {
+        name: 'Changping Chen',
+      },
+    };
+
+    commentsQuery = gql`
+      query Comment($repoName: String!) {
+        entry(repoFullName: $repoName) {
+          comments {
+            text
+          }
+        }
+      }
+    `;
+
+    commentsSub = gql`
+      subscription getNewestComment($repoName: String!) {
+        getNewestComment(repoName: $repoName) {
+          text
+        }
+      }`;
+
+    commentsVariables = {
+      repoName: 'org/repo',
+    };
+
+    commentsWatchQueryOptions = {
+      query: commentsQuery,
+      variables: commentsVariables,
+    };
+
+    commentsResult = {
+      data: {
+        entry: {
+          comments: [],
+        },
+      },
+    };
+
+    commentsResultMore = {
+      result: {
+        entry: {
+          comments: [],
+        },
+      },
+    };
+
+    for (let i = 1; i <= 10; i++) {
+      commentsResult.data.entry.comments.push({ text: `comment ${i}` });
+    }
+
+    for (let i = 11; i < 12; i++) {
+      commentsResultMore.result.entry.comments.push({ text: `comment ${i}` });
+    }
+
+    sub2 = {
+      request: {
+        query: commentsSub,
+        variables: commentsVariables,
+      },
+      id: 0,
+      results: [commentsResultMore],
+    };
+
+  });
+
+
+
+
+  it('should start a subscription on network interface', (done) => {
+    const network = mockSubscriptionNetworkInterface([sub1]);
+    const queryManager = new QueryManager({
+      networkInterface: network,
+      reduxRootKey: 'apollo',
+      store: createApolloStore(),
+    });
+    options.handler = (error, result) => {
+      assert.deepEqual(result, results[0].result);
+      done();
+    };
+    const id = queryManager.startSubscription(options);
+    network.fireResult(id);
+  });
+
+  it('should receive multiple results for a subscription', (done) => {
+    const network = mockSubscriptionNetworkInterface([sub1]);
+    let numResults = 0;
+    const queryManager = new QueryManager({
+      networkInterface: network,
+      reduxRootKey: 'apollo',
+      store: createApolloStore(),
+    });
+    options.handler = (error, result) => {
+      assert.deepEqual(result, results[numResults].result);
+      numResults++;
+      if (numResults === 4) {
+        done();
+      }
+    };
+    const id = queryManager.startSubscription(options);
+    for (let i = 0; i < 4; i++) {
+      network.fireResult(id);
+    }
+  });
+
+  it('should work with an observable query', (done) => {
+    const network = mockSubscriptionNetworkInterface([sub2], {
+      request: {
+        query: commentsQuery,
+        variables: commentsVariables,
+      },
+      result: commentsResult, // list of 10 comments
+    });
+    const client = new ApolloClient({
+      networkInterface: network,
+    });
+    client.query({
+      query: commentsQuery,
+      variables: commentsVariables,
+    }).then(() => {
+      const graphQLSubscriptionOptions = {
+        subscription: commentsSub,
+        variables: commentsVariables,
+        updateFunction: (prev, updateOptions) => {
+          const state = clonedeep(prev) as any;
+          // prev is that data field of the query result
+          // updateOptions.subscriptionResult is the result entry from the subscription result
+          state.entry.comments = [...state.entry.comments, ...(updateOptions.subscriptionResult as any).entry.comments];
+          return state;
+        },
+      };
+      const obsHandle = client.watchQuery(commentsWatchQueryOptions);
+
+      obsHandle.subscribe({
+        next(result) {
+          let expectedComments = [];
+          for (let i = 1; i <= 11; i++) {
+            expectedComments.push({ text: `comment ${i}` });
+          }
+          assert.equal(result.data.entry.comments.length, 11);
+          assert.deepEqual(result.data.entry.comments, expectedComments);
+          done();
+        },
+      });
+
+      const id = obsHandle.startGraphQLSubscription(graphQLSubscriptionOptions);
+      network.fireResult(id);
+    });
+  });
+
+  // TODO: test that we can make two subscriptions one one watchquery.
+
+});