2.x: fix switchMap bad cancellation (#4513)

Author: akarnokd

Diff for: src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchMap.java

@@ -175,6 +175,7 @@ public void request(long n) {
         public void cancel() {
             if (!cancelled) {
                 cancelled = true;
+                s.cancel();
 
                 disposeInner();
             }
@@ -186,7 +187,7 @@ void disposeInner() {
             if (a != CANCELLED) {
                 a = active.getAndSet((SwitchMapInnerSubscriber<T, R>)CANCELLED);
                 if (a != CANCELLED && a != null) {
-                    s.cancel();
+                    a.cancel();
                 }
             }
         }

Diff for: src/main/java/io/reactivex/internal/operators/observable/ObservableSwitchMap.java

@@ -160,7 +160,7 @@ public void onComplete() {
         public void dispose() {
             if (!cancelled) {
                 cancelled = true;
-
+                s.dispose();
                 disposeInner();
             }
         }
@@ -176,7 +176,7 @@ void disposeInner() {
             if (a != CANCELLED) {
                 a = active.getAndSet((SwitchMapInnerSubscriber<T, R>)CANCELLED);
                 if (a != CANCELLED && a != null) {
-                    s.dispose();
+                    a.cancel();
                 }
             }
         }

Diff for: src/test/java/io/reactivex/internal/operators/flowable/FlowableSwitchTest.java

@@ -797,4 +797,18 @@ public Publisher<Integer> apply(Object v) throws Exception {
 
     }
 
+    @Test
+    public void switchMapInnerCancelled() {
+        PublishProcessor<Integer> pp = PublishProcessor.create();
+
+        TestSubscriber<Integer> ts = Flowable.just(1)
+                .switchMap(Functions.justFunction(pp))
+                .test();
+
+        assertTrue(pp.hasSubscribers());
+
+        ts.cancel();
+
+        assertFalse(pp.hasSubscribers());
+    }
 }

Diff for: src/test/java/io/reactivex/internal/operators/observable/ObservableSwitchTest.java

@@ -607,4 +607,20 @@ public ObservableSource<Integer> apply(Object v) throws Exception {
 
     }
 
+
+    @Test
+    public void switchMapInnerCancelled() {
+        PublishSubject<Integer> pp = PublishSubject.create();
+
+        TestObserver<Integer> ts = Observable.just(1)
+                .switchMap(Functions.justFunction(pp))
+                .test();
+
+        assertTrue(pp.hasObservers());
+
+        ts.cancel();
+
+        assertFalse(pp.hasObservers());
+    }
+
 }