Fix cleanup observer blocking unsubscribe (#7165)

Fixes #6985.

CHANGELOG.md

@@ -18,6 +18,9 @@
   [@hwillson](https://github.com/hwillson) in [#7110](https://github.com/apollographql/apollo-client/pull/7110)
   > Returning mock configuration errors through the Link's `Observable` was the default behavior in Apollo Client 2.x. We changed it for 3, but the change has been problematic for those looking to migrate from 2.x to 3. We've decided to change this back with the understanding that not many people want or are relying on `MockLink`'s throwing exception approach. If you want to change this functionality, you can define custom error handling through `MockLink.setOnError`.
 
+- Unsubscribing the last observer from an `ObservableQuery` will once again unsubscribe from the underlying network `Observable` in all cases, as in Apollo Client 2.x, allowing network requests to be cancelled by unsubscribing. <br/>
+  [@javier-garcia-meteologica](https://github.com/javier-garcia-meteologica) in [#7165](https://github.com/apollographql/apollo-client/pull/7165)
+
 ## Improvements
 
 - Support inheritance of type and field policies, according to `possibleTypes`. <br/>

src/utilities/observables/Concast.ts

@@ -95,26 +95,34 @@ export class Concast<T> extends Observable<T> {
     this.handlers.complete!();
   }
 
+  private deliverLastMessage(observer: Observer<T>) {
+    if (this.latest) {
+      const nextOrError = this.latest[0];
+      const method = observer[nextOrError];
+      if (method) {
+        method.call(observer, this.latest[1]);
+      }
+      // If the subscription is already closed, and the last message was
+      // a 'next' message, simulate delivery of the final 'complete'
+      // message again.
+      if (this.sub === null &&
+          nextOrError === "next" &&
+          observer.complete) {
+        observer.complete();
+      }
+    }
+  }
+
+  // Note: cleanup observers do not count towards this total.
+  private addCount = 0;
+
   public addObserver(observer: Observer<T>) {
     if (!this.observers.has(observer)) {
       // Immediately deliver the most recent message, so we can always
       // be sure all observers have the latest information.
-      if (this.latest) {
-        const nextOrError = this.latest[0];
-        const method = observer[nextOrError];
-        if (method) {
-          method.call(observer, this.latest[1]);
-        }
-        // If the subscription is already closed, and the last message was
-        // a 'next' message, simulate delivery of the final 'complete'
-        // message again.
-        if (this.sub === null &&
-            nextOrError === "next" &&
-            observer.complete) {
-          observer.complete();
-        }
-      }
+      this.deliverLastMessage(observer);
       this.observers.add(observer);
+      ++this.addCount;
     }
   }
 
@@ -201,7 +209,13 @@ export class Concast<T> extends Observable<T> {
     const once = () => {
       if (!called) {
         called = true;
-        this.observers.delete(observer);
+        // If there have been no other (non-cleanup) observers added, pass
+        // true for the quietly argument, so the removal of the cleanup
+        // observer does not call this.sub.unsubscribe. If a cleanup
+        // observer is added and removed before any other observers
+        // subscribe, we do not want to prevent other observers from
+        // subscribing later.
+        this.removeObserver(observer, !this.addCount);
         callback();
       }
     }
@@ -210,7 +224,12 @@ export class Concast<T> extends Observable<T> {
       error: once,
       complete: once,
     };
+    const count = this.addCount;
     this.addObserver(observer);
+    // Normally addObserver increments this.addCount, but we can "hide"
+    // cleanup observers by restoring this.addCount to its previous value
+    // after adding any cleanup observer.
+    this.addCount = count;
   }
 
   // A public way to abort observation and broadcast.