subscribeToMore for observableQuery

CHANGELOG.md

@@ -24,6 +24,7 @@ Expect active development and potentially significant breaking changes in the `0
 - Fix multidimentional array handling. [Issue #776](https://github.com/apollostack/apollo-client/issues/776) [PR #785](https://github.com/apollostack/apollo-client/pull/785)
 - Add support for Enum inline arguments [Issue #183](https://github.com/apollostack/apollo-client/issues/183) [PR #788](https://github.com/apollostack/apollo-client/pull/788)
 - Make it possible to subscribe to the same observable query multiple times. The query is initialized on the first subscription, and torn down after the last. Now, QueryManager is only aware of one subscription from the ObservableQuery, no matter how many were actually registered. This fixes issues with `result()` and other ObservableQuery features that relied on subscribing multiple times to work. This should remove the need for the workaround in `0.4.21`. [Repro in PR #694](https://github.com/apollostack/apollo-client/pull/694) [PR #791](https://github.com/apollostack/apollo-client/pull/791)
+- ** new Feature **: Add fetchMore-style subscribeToMore function which updates a query result based on a subscription. [PR #797](https://github.com/apollostack/apollo-client/pull/797)
 
 ### v0.4.21
 - Added some temporary functions (`_setVariablesNoResult` and `_setOptionsNoResult`) to work around a `react-apollo` problem fundamentally caused by the issue highlighted in [PR #694](https://github.com/apollostack/apollo-client/pull/694). The code as been refactored on `master`, so we expect it to be fixed in 0.5.x, and is not worth resolving now.

src/core/ObservableQuery.ts

@@ -2,9 +2,10 @@ import {
   ModifiableWatchQueryOptions,
   WatchQueryOptions,
   FetchMoreQueryOptions,
+  SubscribeToMoreOptions,
 } from './watchQueryOptions';
 
-import { Observable, Observer } from '../util/Observable';
+import { Observable, Observer, Subscription } from '../util/Observable';
 
 import {
   QueryScheduler,
@@ -49,6 +50,7 @@ export class ObservableQuery extends Observable<ApolloQueryResult> {
   private scheduler: QueryScheduler;
   private queryManager: QueryManager;
   private observers: Observer<ApolloQueryResult>[];
+  private subscriptionHandles: Subscription[];
 
   private lastResult: ApolloQueryResult;
   private lastError: ApolloError;
@@ -80,6 +82,7 @@ export class ObservableQuery extends Observable<ApolloQueryResult> {
     this.queryId = queryId;
     this.shouldSubscribe = shouldSubscribe;
     this.observers = [];
+    this.subscriptionHandles = [];
   }
 
   public result(): Promise<ApolloQueryResult> {
@@ -165,7 +168,8 @@ export class ObservableQuery extends Observable<ApolloQueryResult> {
         const reducer = fetchMoreOptions.updateQuery;
         const mapFn = (previousResult: any, { variables }: {variables: any }) => {
 
-          // TODO REFACTOR: reached max recursion depth (figuratively). Continue renaming to variables further down when we have time.
+          // TODO REF: reached max recursion depth (fig) when renaming queryVariables to variables.
+          // Continue renaming to variables further down when we have time.
           const queryVariables = variables;
           return reducer(
             previousResult, {
@@ -178,6 +182,49 @@ export class ObservableQuery extends Observable<ApolloQueryResult> {
       });
   }
 
+  // XXX the subscription variables are separate from the query variables.
+  // if you want to update subscription variables, right now you have to do that separately,
+  // and you can only do it by stopping the subscription and then subscribing again with new variables.
+  public subscribeToMore(
+    options: SubscribeToMoreOptions,
+  ): () => void {
+    const observable = this.queryManager.startGraphQLSubscription({
+      document: options.document,
+      variables: options.variables,
+    });
+
+    const reducer = options.updateQuery;
+
+    const subscription = observable.subscribe({
+      next: (subscriptionData) => {
+        const mapFn = (previousResult: Object, { variables }: { variables: Object }) => {
+          return reducer(
+            previousResult, {
+              subscriptionData,
+              variables,
+            }
+          );
+        };
+        this.updateQuery(mapFn);
+      },
+      error: (err) => {
+        // TODO implement something smart here when improving error handling
+        console.error(err);
+      },
+    });
+
+    this.subscriptionHandles.push(subscription);
+
+    return () => {
+      // XXX technically we should also remove it from this.subscriptionHandles
+      const i = this.subscriptionHandles.indexOf(subscription);
+      if (i >= 0) {
+        this.subscriptionHandles.splice(i, 1);
+        subscription.unsubscribe();
+      }
+    };
+  }
+
   public setOptions(opts: ModifiableWatchQueryOptions): Promise<ApolloQueryResult> {
     const oldOptions = this.options;
     this.options = assign({}, this.options, opts) as WatchQueryOptions;
@@ -342,6 +389,9 @@ export class ObservableQuery extends Observable<ApolloQueryResult> {
       this.scheduler.stopPollingQuery(this.queryId);
     }
 
+    // stop all active GraphQL subscriptions
+    this.subscriptionHandles.forEach( sub => sub.unsubscribe() );
+
     this.queryManager.stopQuery(this.queryId);
     this.observers = [];
   }

src/core/watchQueryOptions.ts

@@ -80,6 +80,15 @@ export interface FetchMoreQueryOptions {
   variables?: { [key: string]: any };
 }
 
+export type SubscribeToMoreOptions = {
+  document: Document;
+  variables?: { [key: string]: any };
+  updateQuery: (previousQueryResult: Object, options: {
+    subscriptionData: any,
+    variables: { [key: string]: any },
+  }) => Object;
+}
+
 export interface DeprecatedSubscriptionOptions {
   query: Document;
   variables?: { [key: string]: any };

test/subscribeToMore.ts

@@ -0,0 +1,92 @@
+import * as chai from 'chai';
+const { assert } = chai;
+
+import {
+  mockSubscriptionNetworkInterface,
+} from './mocks/mockNetworkInterface';
+import ApolloClient from '../src';
+
+// import assign = require('lodash.assign');
+// import clonedeep = require('lodash.clonedeep');
+
+import gql from 'graphql-tag';
+
+describe('subscribeToMore', () => {
+  const query = gql`
+    query aQuery {
+      entry {
+        value
+      }
+    }
+  `;
+  const result = {
+    data: {
+      entry: {
+        value: 1,
+      },
+    },
+  };
+
+  const req1 = { request: { query }, result };
+
+  const results = ['Dahivat Pandya', 'Amanda Liu'].map(
+    name => ({ result: { name: name }, delay: 10 })
+  );
+
+  const sub1 = {
+    request: {
+      query: gql`
+        subscription newValues {
+          name
+        }
+      `,
+    },
+    id: 0,
+    results: [...results],
+  };
+
+  it('triggers new result from subscription data', (done) => {
+    let latestResult: any = null;
+    const networkInterface = mockSubscriptionNetworkInterface([sub1], req1);
+    let counter = 0;
+
+    const client = new ApolloClient({
+      networkInterface,
+      addTypename: false,
+    });
+
+    const obsHandle = client.watchQuery({
+      query,
+    });
+    const sub = obsHandle.subscribe({
+      next(queryResult) {
+        latestResult = queryResult;
+        counter++;
+      },
+    });
+
+    obsHandle.subscribeToMore({
+      document: gql`
+        subscription newValues {
+          name
+        }
+      `,
+      updateQuery: (prev, { subscriptionData }) => {
+        return { entry: { value: subscriptionData.name } };
+      },
+    });
+
+    setTimeout(() => {
+      sub.unsubscribe();
+      assert.equal(counter, 3);
+      assert.deepEqual(latestResult, { data: { entry: { value: 'Amanda Liu' } }, loading: false });
+      done();
+    }, 50);
+
+    for (let i = 0; i < 2; i++) {
+      networkInterface.fireResult(0); // 0 is the id of the subscription for the NI
+    }
+  });
+
+  // TODO add a test that checks that subscriptions are cancelled when obs is unsubscribed from.
+});

test/tests.ts

@@ -40,3 +40,4 @@ import './mockNetworkInterface';
 import './graphqlSubscriptions';
 import './batchedNetworkInterface';
 import './ObservableQuery';
+import './subscribeToMore';