Fixed single response and added subscriptions to links (apollographql…

…#1992)

* Fixed single response and added subscriptions to links

* remove Observable Network Interface and include subscriptions map for current interface

* Subscripitons unit tests and subscribe method calls correctly to handler

* Update Changelog to reflect subscriptions with Links

* bumped graphql types version and include asynciterable ts library

.vscode/launch.json

@@ -4,6 +4,7 @@
     {
       "name": "Test",
       "type": "node",
+      "request": "launch",
       "program": "${workspaceRoot}/node_modules/mocha/bin/_mocha",
       "stopOnEntry": false,
       "args": [

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### vNEXT
 
+-Add support for subscriptions with Apollo Link network stack [PR #1992](https://github.com/apollographql/apollo-client/pull/1992)
 -fixed `resolved` scoping issue for multiple queries in flight with Apollo Link [PR #2002](https://github.com/apollographql/apollo-client/pull/2002)
 
 ### 1.9.0

package.json

@@ -128,7 +128,6 @@
     "webpack-bundle-analyzer": "2.9.0"
   },
   "optionalDependencies": {
-    "@types/async": "2.0.40",
-    "@types/graphql": "0.9.4"
+    "@types/graphql": "0.10.2"
   }
 }

src/ApolloClient.ts

@@ -5,7 +5,12 @@ import {
   Request,
 } from './transport/networkInterface';
 
-import { execute, ApolloLink } from 'apollo-link-core';
+import {
+  execute,
+  ApolloLink,
+  FetchResult,
+  Observable as ZenObservable,
+} from 'apollo-link-core';
 import { assign } from './util/assign';
 
 import {
@@ -131,6 +136,7 @@ export default class ApolloClient implements DataProxy {
   private proxy: DataProxy | undefined;
   private fragmentMatcher: FragmentMatcherInterface;
   private ssrMode: boolean;
+  private subscriptionMap: Map<string, ZenObservable.Subscription>;
 
   /**
    * Constructs an instance of {@link ApolloClient}.
@@ -229,25 +235,45 @@ export default class ApolloClient implements DataProxy {
     };
 
     if (networkInterface instanceof ApolloLink) {
+      let count = 0;
       this.networkInterface = {
         query: createQuery((request: Request) => {
           return (execute(
             networkInterface as ApolloLink,
             request,
           ) as any) as Observable<ExecutionResult>;
         }),
+        subscribe: (request: any, handler: any): string => {
+          if (!this.subscriptionMap) {
+            this.subscriptionMap = new Map<
+              string,
+              ZenObservable.Subscription
+            >();
+          }
+
+          const subscription = (execute(
+            networkInterface as ApolloLink,
+            request,
+          ) as any).subscribe({
+            next: (data: FetchResult) => handler(undefined, data),
+            error: (error: Error) => handler([error]),
+            complete: handler,
+          });
+
+          const id = count.toString();
+          this.subscriptionMap.set(id, subscription);
+          count++;
+          return id;
+        },
+        unsubscribe: (id: string): void => {
+          if (this.subscriptionMap) {
+            const subscription = this.subscriptionMap.get(id);
+            if (subscription) {
+              subscription.unsubscribe();
+            }
+          }
+        },
       };
-    } else if (
-      networkInterface &&
-      typeof (<ObservableNetworkInterface>networkInterface).request ===
-        'function'
-    ) {
-      console.warn(`The Observable Network interface will be deprecated`);
-      this.networkInterface = assign(networkInterface || {}, {
-        query: createQuery(
-          (networkInterface as ObservableNetworkInterface).request,
-        ),
-      });
     } else {
       this.networkInterface = networkInterface
         ? <NetworkInterface>networkInterface

test/graphqlSubscriptions.ts

@@ -17,6 +17,10 @@ import { QueryManager } from '../src/core/QueryManager';
 
 import { createApolloStore } from '../src/store';
 
+import { SubscriptionOptions } from '../src/core/watchQueryOptions';
+
+import { ApolloLink, Observable, FetchResult } from 'apollo-link-core';
+
 describe('GraphQL Subscriptions', () => {
   const results = [
     'Dahivat Pandya',
@@ -205,6 +209,83 @@ describe('GraphQL Subscriptions', () => {
     }
   });
 
+  it('should receive multiple results for a subscription with Apollo Link', done => {
+    let numResults = 0;
+    const expectedData = [
+      'Dahivat Pandya',
+      'Vyacheslav Kim',
+      'Changping Chen',
+      'Amanda Liu',
+    ].map(name => ({ user: { name: name } }));
+    const queryInfo = {
+      request: defaultSub1.request || ({} as any),
+      results: [...expectedData],
+    };
+
+    const link = ApolloLink.from([
+      operation =>
+        new Observable(observer => {
+          if (queryInfo.results) {
+            queryInfo.results.map(result =>
+              observer.next(result as FetchResult),
+            );
+          }
+        }),
+    ]);
+    const client = new ApolloClient({
+      networkInterface: link,
+      addTypename: false,
+    });
+
+    const obs = client.subscribe(queryInfo.request as SubscriptionOptions);
+    const sub = obs.subscribe({
+      next: data => {
+        const expected = expectedData.shift();
+        if (expected) {
+          assert.equal(data, expected);
+        } else {
+          assert(false);
+        }
+        if (expectedData.length === 0) {
+          done();
+        }
+      },
+      error: console.log,
+      complete: () => assert(false),
+    });
+  });
+
+  it('should unsubscribe properly with Apollo Link', () => {
+    let numResults = 0;
+    const readyToUnsub: boolean[] = [];
+
+    const link = ApolloLink.from([
+      operation =>
+        new Observable(observer => {
+          const index = readyToUnsub.length;
+          readyToUnsub.push(false);
+          return () => {
+            assert(readyToUnsub[index]);
+          };
+        }),
+    ]);
+    const client = new ApolloClient({
+      networkInterface: link,
+      addTypename: false,
+    });
+
+    const obs0 = client.subscribe(defaultSub1.request as SubscriptionOptions);
+    const obs1 = client.subscribe(defaultSub1.request as SubscriptionOptions);
+    const subscription0 = obs0.subscribe({});
+    const subscription1 = obs1.subscribe({});
+
+    readyToUnsub[1] = true;
+    subscription1.unsubscribe();
+    readyToUnsub[0] = true;
+    subscription0.unsubscribe();
+    readyToUnsub.map(bool => assert(bool));
+  });
+
   it('should fire redux action and call result reducers', done => {
     const query = gql`
       query miniQuery {

tsconfig.json

@@ -17,6 +17,7 @@
   },
   "files": [
     "node_modules/typescript/lib/lib.es2015.d.ts",
+    "node_modules/typescript/lib/lib.esnext.asynciterable.d.ts",
     "node_modules/typescript/lib/lib.dom.d.ts",
     "typings.d.ts",
     "fetch-mock.typings.d.ts",