Fixes the DedupLink shared observable unsubscribing early when shared

CHANGELOG.md

@@ -2,6 +2,17 @@
 
 ## vNEXT
 
+### apollo-link-dedup 1.0.17
+
+- Fixes an issue caused by the `DedupLink` shared observable returning 
+  cleanup logic that unsubscribes from the real observable, without 
+  checking whether only one of the many (shared) subscribers are 
+  unsubscribing. This caused problems when using `DedupLink` in front of 
+  `HttpLink`, as this lead to `HttpLink` aborting HTTP requests while some 
+  callers were still waiting for a response.  <br/>
+  [@ms](https://github.com/ms) in [#984](https://github.com/apollographql/apollo-link/pull/984)
+
+
 ## 2019-03-05
 
 ### General

packages/apollo-link-dedup/src/__tests__/dedupLink.ts

@@ -142,6 +142,7 @@ describe('DedupLink', () => {
     execute(deduper, request2).subscribe({});
     expect(called).toBe(1);
   });
+
   it(`works for nested queries`, done => {
     const document: DocumentNode = gql`
       query test1($x: String) {
@@ -226,6 +227,7 @@ describe('DedupLink', () => {
     execute(deduper, request2).subscribe({});
     expect(called).toBe(2);
   });
+
   it(`unsubscribes as needed`, () => {
     const document: DocumentNode = gql`
       query test1($x: String) {
@@ -267,4 +269,46 @@ describe('DedupLink', () => {
 
     expect(unsubscribed).toBe(true);
   });
+
+  it(`unsubscribes only when needed`, () => {
+    const document: DocumentNode = gql`
+      query test1($x: String) {
+        test(x: $x)
+      }
+    `;
+    const variables1 = { x: 'Hello World' };
+    const variables2 = { x: 'Hello World' };
+
+    const request1: GraphQLRequest = {
+      query: document,
+      variables: variables1,
+      operationName: getOperationName(document),
+    };
+
+    const request2: GraphQLRequest = {
+      query: document,
+      variables: variables2,
+      operationName: getOperationName(document),
+    };
+
+    let unsubscribed = false;
+    const deduper = ApolloLink.from([
+      new DedupLink(),
+      new ApolloLink(() => {
+        return new Observable(() => {
+          return () => {
+            unsubscribed = true;
+          };
+        });
+      }),
+    ]);
+
+    const sub1 = execute(deduper, request1).subscribe({});
+    const sub2 = execute(deduper, request2).subscribe({});
+
+    sub1.unsubscribe();
+
+    //NOTE: sub2 is still waiting!
+    expect(unsubscribed).toBe(false);
+  });
 });

packages/apollo-link-dedup/src/dedupLink.ts

@@ -27,12 +27,6 @@ export class DedupLink extends ApolloLink {
 
     const key = operation.toKey();
 
-    const cleanup = operationKey => {
-      this.inFlightRequestObservables.delete(operationKey);
-      const prev = this.subscribers.get(operationKey);
-      return prev;
-    };
-
     if (!this.inFlightRequestObservables.get(key)) {
       // this is a new request, i.e. we haven't deduplicated it yet
       // call the next link
@@ -42,36 +36,38 @@ export class DedupLink extends ApolloLink {
       const sharedObserver = new Observable(observer => {
         // this will still be called by each subscriber regardless of
         // deduplication status
-        let prev = this.subscribers.get(key);
-        if (!prev) prev = { next: [], error: [], complete: [] };
+        if (!this.subscribers.has(key)) this.subscribers.set(key, new Set());
 
-        this.subscribers.set(key, {
-          next: prev.next.concat([observer.next.bind(observer)]),
-          error: prev.error.concat([observer.error.bind(observer)]),
-          complete: prev.complete.concat([observer.complete.bind(observer)]),
-        });
+        this.subscribers.get(key).add(observer);
 
         if (!subscription) {
           subscription = singleObserver.subscribe({
             next: result => {
-              const previous = cleanup(key);
+              const subscribers = this.subscribers.get(key);
               this.subscribers.delete(key);
-              if (previous) {
-                previous.next.forEach(next => next(result));
-                previous.complete.forEach(complete => complete());
+              this.inFlightRequestObservables.delete(key);
+              if (subscribers) {
+                subscribers.forEach(obs => obs.next(result));
+                subscribers.forEach(obs => obs.complete());
               }
             },
             error: error => {
-              const previous = cleanup(key);
+              const subscribers = this.subscribers.get(key);
               this.subscribers.delete(key);
-              if (previous) previous.error.forEach(err => err(error));
+              this.inFlightRequestObservables.delete(key);
+              if (subscribers) {
+                subscribers.forEach(obs => obs.error(error));
+              }
             },
           });
         }
 
         return () => {
-          if (subscription) subscription.unsubscribe();
-          this.inFlightRequestObservables.delete(key);
+          this.subscribers.get(key).delete(observer);
+          if (this.subscribers.get(key).size === 0) {
+            this.inFlightRequestObservables.delete(key);
+            if (subscription) subscription.unsubscribe();
+          }
         };
       });