make subscriptions fire redux action

src/actions.ts

@@ -134,6 +134,20 @@ export function isStoreResetAction(action: ApolloAction): action is StoreResetAc
   return action.type === 'APOLLO_STORE_RESET';
 }
 
+export type SubscriptionResultAction = {
+  type: 'APOLLO_SUBSCRIPTION_RESULT';
+  result: GraphQLResult;
+  subscriptionId: number;
+  variables: Object;
+  document: Document;
+  operationName: string;
+  extraReducers?: ApolloReducer[];
+}
+
+export function isSubscriptionResultAction(action: ApolloAction): action is SubscriptionResultAction {
+  return action.type === 'APOLLO_SUBSCRIPTION_RESULT';
+}
+
 export type ApolloAction =
   QueryResultAction |
   QueryErrorAction |
@@ -144,4 +158,5 @@ export type ApolloAction =
   MutationResultAction |
   MutationErrorAction |
   UpdateQueryResultAction |
-  StoreResetAction;
+  StoreResetAction |
+  SubscriptionResultAction;

src/core/QueryManager.ts

@@ -560,15 +560,15 @@ export class QueryManager {
       document,
       variables,
     } = options;
-    let queryDoc = document;
+    let transformedDoc = document;
     // Apply the query transformer if one has been provided.
     if (this.queryTransformer) {
-      queryDoc = applyTransformers(queryDoc, [this.queryTransformer]);
+      transformedDoc = applyTransformers(transformedDoc, [this.queryTransformer]);
     }
     const request: Request = {
-      query: queryDoc,
+      query: transformedDoc,
       variables,
-      operationName: getOperationName(queryDoc),
+      operationName: getOperationName(transformedDoc),
     };
 
     let subId: number;
@@ -577,6 +577,8 @@ export class QueryManager {
     return new Observable((observer) => {
       observers.push(observer);
 
+      // TODO REFACTOR: the result here is not a normal GraphQL result.
+
       // If this is the first observer, actually initiate the network subscription
       if (observers.length === 1) {
         const handler = (error: Error, result: any) => {
@@ -585,6 +587,16 @@ export class QueryManager {
               obs.error(error);
             });
           } else {
+            this.store.dispatch({
+              type: 'APOLLO_SUBSCRIPTION_RESULT',
+              document: transformedDoc,
+              operationName: getOperationName(transformedDoc),
+              result: { data: result },
+              variables,
+              subscriptionId: subId,
+              extraReducers: this.getExtraReducers(),
+            });
+            // It's slightly awkward that the data for subscriptions doesn't come from the store.
             observers.forEach((obs) => {
               obs.next(result);
             });
@@ -771,6 +783,21 @@ export class QueryManager {
     };
   }
 
+  private getExtraReducers() {
+    return  Object.keys(this.observableQueries).map( obsQueryId => {
+      const queryOptions = this.observableQueries[obsQueryId].observableQuery.options;
+      if (queryOptions.reducer) {
+        return createStoreReducer(
+          queryOptions.reducer,
+          queryOptions.query,
+          queryOptions.variables,
+          this.reducerConfig,
+          );
+      }
+      return null;
+    }).filter( reducer => reducer !== null );
+  }
+
   // Takes a request id, query id, a query document and information associated with the query
   // and send it to the network interface. Returns
   // a promise for the result associated with that request.
@@ -802,18 +829,7 @@ export class QueryManager {
       return this.networkInterface.query(request)
         .then((result: GraphQLResult) => {
 
-          const extraReducers = Object.keys(this.observableQueries).map( obsQueryId => {
-            const queryOptions = this.observableQueries[obsQueryId].observableQuery.options;
-            if (queryOptions.reducer) {
-              return createStoreReducer(
-                queryOptions.reducer,
-                queryOptions.query,
-                queryOptions.variables,
-                this.reducerConfig,
-                );
-            }
-            return null;
-          }).filter( reducer => reducer !== null );
+          const extraReducers = this.getExtraReducers();
 
           // XXX handle multiple ApolloQueryResults
           this.store.dispatch({

src/data/store.ts

@@ -4,6 +4,7 @@ import {
   isMutationResultAction,
   isUpdateQueryResultAction,
   isStoreResetAction,
+  isSubscriptionResultAction,
 } from '../actions';
 
 import {
@@ -121,6 +122,35 @@ export function data(
         });
       }
 
+      return newState;
+    }
+  } else if (isSubscriptionResultAction(action)) {
+    // the subscription interface should handle not sending us results we no longer subscribe to.
+    // XXX I don't think we ever send in an object with errors, but we might in the future...
+    if (! graphQLResultHasError(action.result)) {
+
+      // XXX use immutablejs instead of cloning
+      const clonedState = assign({}, previousState) as NormalizedCache;
+
+      // TODO REFACTOR: is writeResultToStore a good name for something that doesn't actually
+      // write to "the" store?
+      let newState = writeResultToStore({
+        result: action.result.data,
+        dataId: 'ROOT_QUERY', // TODO: is this correct? what am I doing here? What is dataId for??
+        document: action.document,
+        variables: action.variables,
+        store: clonedState,
+        dataIdFromObject: config.dataIdFromObject,
+      });
+
+      // XXX each reducer gets the state from the previous reducer.
+      // Maybe they should all get a clone instead and then compare at the end to make sure it's consistent.
+      if (action.extraReducers) {
+        action.extraReducers.forEach( reducer => {
+          newState = reducer(newState, constAction);
+        });
+      }
+
       return newState;
     }
   } else if (isMutationResultAction(constAction)) {