3.x: fix switchMaps inconsistency swallowing errors when cancelled

src/main/java/io/reactivex/internal/operators/flowable/FlowableFlatMap.java

@@ -585,10 +585,7 @@ void disposeAll() {
                     for (InnerSubscriber<?, ?> inner : a) {
                         inner.dispose();
                     }
-                    Throwable ex = errs.terminate();
-                    if (ex != null && ex != ExceptionHelper.TERMINATED) {
-                        RxJavaPlugins.onError(ex);
-                    }
+                    errs.tryTerminateAndReport();
                 }
             }
         }

src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchMap.java

@@ -173,6 +173,8 @@ public void cancel() {
                 upstream.cancel();
 
                 disposeInner();
+
+                error.tryTerminateAndReport();
             }
         }
 

src/main/java/io/reactivex/internal/operators/mixed/FlowableSwitchMapCompletable.java

@@ -162,6 +162,7 @@ void disposeInner() {
         public void dispose() {
             upstream.cancel();
             disposeInner();
+            errors.tryTerminateAndReport();
         }
 
         @Override
@@ -178,7 +179,8 @@ void innerError(SwitchMapInnerObserver sender, Throwable error) {
                             downstream.onError(ex);
                         }
                     } else {
-                        dispose();
+                        upstream.cancel();
+                        disposeInner();
                         Throwable ex = errors.terminate();
                         if (ex != ExceptionHelper.TERMINATED) {
                             downstream.onError(ex);

src/main/java/io/reactivex/internal/operators/mixed/FlowableSwitchMapMaybe.java

@@ -177,6 +177,7 @@ public void cancel() {
             cancelled = true;
             upstream.cancel();
             disposeInner();
+            errors.tryTerminateAndReport();
         }
 
         void innerError(SwitchMapMaybeObserver<R> sender, Throwable ex) {

src/main/java/io/reactivex/internal/operators/mixed/FlowableSwitchMapSingle.java

@@ -177,6 +177,7 @@ public void cancel() {
             cancelled = true;
             upstream.cancel();
             disposeInner();
+            errors.tryTerminateAndReport();
         }
 
         void innerError(SwitchMapSingleObserver<R> sender, Throwable ex) {

src/main/java/io/reactivex/internal/operators/mixed/ObservableSwitchMapCompletable.java

@@ -160,6 +160,7 @@ void disposeInner() {
         public void dispose() {
             upstream.dispose();
             disposeInner();
+            errors.tryTerminateAndReport();
         }
 
         @Override
@@ -176,7 +177,8 @@ void innerError(SwitchMapInnerObserver sender, Throwable error) {
                             downstream.onError(ex);
                         }
                     } else {
-                        dispose();
+                        upstream.dispose();
+                        disposeInner();
                         Throwable ex = errors.terminate();
                         if (ex != ExceptionHelper.TERMINATED) {
                             downstream.onError(ex);

src/main/java/io/reactivex/internal/operators/mixed/ObservableSwitchMapMaybe.java

@@ -164,6 +164,7 @@ public void dispose() {
             cancelled = true;
             upstream.dispose();
             disposeInner();
+            errors.tryTerminateAndReport();
         }
 
         @Override

src/main/java/io/reactivex/internal/operators/mixed/ObservableSwitchMapSingle.java

@@ -164,6 +164,7 @@ public void dispose() {
             cancelled = true;
             upstream.dispose();
             disposeInner();
+            errors.tryTerminateAndReport();
         }
 
         @Override

src/main/java/io/reactivex/internal/operators/observable/ObservableFlatMap.java

@@ -307,10 +307,7 @@ public void dispose() {
             if (!cancelled) {
                 cancelled = true;
                 if (disposeAll()) {
-                    Throwable ex = errors.terminate();
-                    if (ex != null && ex != ExceptionHelper.TERMINATED) {
-                        RxJavaPlugins.onError(ex);
-                    }
+                    errors.tryTerminateAndReport();
                 }
             }
         }

src/main/java/io/reactivex/internal/operators/observable/ObservableSwitchMap.java

@@ -157,6 +157,8 @@ public void dispose() {
                 cancelled = true;
                 upstream.dispose();
                 disposeInner();
+
+                errors.tryTerminateAndReport();
             }
         }
 

src/main/java/io/reactivex/internal/util/AtomicThrowable.java

@@ -15,6 +15,8 @@
 
 import java.util.concurrent.atomic.AtomicReference;
 
+import io.reactivex.plugins.RxJavaPlugins;
+
 /**
  * Atomic container for Throwables including combining and having a
  * terminal state via ExceptionHelper.
@@ -46,4 +48,17 @@ public Throwable terminate() {
     public boolean isTerminated() {
         return get() == ExceptionHelper.TERMINATED;
     }
+
+    /**
+     * Tries to terminate this atomic throwable (by swapping in the TERMINATED indicator)
+     * and calls {@link RxJavaPlugins#onError(Throwable)} if there was a non-null, non-indicator
+     * exception contained within before.
+     * @since 3.0.0
+     */
+    public void tryTerminateAndReport() {
+        Throwable ex = terminate();
+        if (ex != null && ex != ExceptionHelper.TERMINATED) {
+            RxJavaPlugins.onError(ex);
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/flowable/FlowableSwitchTest.java

@@ -14,6 +14,7 @@
 package io.reactivex.internal.operators.flowable;
 
 import static org.junit.Assert.*;
+import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.*;
@@ -1202,4 +1203,34 @@ public Object apply(Integer w) throws Exception {
         .assertNoErrors()
         .assertComplete();
     }
+
+    @Test
+    public void undeliverableUponCancel() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final TestSubscriberEx<Integer> ts = new TestSubscriberEx<Integer>();
+
+            Flowable.just(1)
+            .map(new Function<Integer, Integer>() {
+                @Override
+                public Integer apply(Integer v) throws Throwable {
+                    ts.cancel();
+                    throw new TestException();
+                }
+            })
+            .switchMap(new Function<Integer, Publisher<Integer>>() {
+                @Override
+                public Publisher<Integer> apply(Integer v) throws Throwable {
+                    return Flowable.just(v).hide();
+                }
+            })
+            .subscribe(ts);
+
+            ts.assertEmpty();
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/mixed/FlowableSwitchMapCompletableTest.java

@@ -29,7 +29,7 @@
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.processors.PublishProcessor;
 import io.reactivex.subjects.CompletableSubject;
-import io.reactivex.testsupport.TestHelper;
+import io.reactivex.testsupport.*;
 
 public class FlowableSwitchMapCompletableTest {
 
@@ -387,4 +387,34 @@ public void mainErrorDelayed() {
 
         to.assertFailure(TestException.class);
     }
+
+    @Test
+    public void undeliverableUponCancel() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final TestObserverEx<Integer> to = new TestObserverEx<Integer>();
+
+            Flowable.just(1)
+            .map(new Function<Integer, Integer>() {
+                @Override
+                public Integer apply(Integer v) throws Throwable {
+                    to.dispose();
+                    throw new TestException();
+                }
+            })
+            .switchMapCompletable(new Function<Integer, Completable>() {
+                @Override
+                public Completable apply(Integer v) throws Throwable {
+                    return Completable.complete().hide();
+                }
+            })
+            .subscribe(to);
+
+            to.assertEmpty();
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/mixed/FlowableSwitchMapMaybeTest.java

@@ -646,4 +646,34 @@ public void onNext(Integer t) {
 
         ts.assertResult(1, 1, 1, 1, 1);
     }
+
+    @Test
+    public void undeliverableUponCancel() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final TestSubscriberEx<Integer> ts = new TestSubscriberEx<Integer>();
+
+            Flowable.just(1)
+            .map(new Function<Integer, Integer>() {
+                @Override
+                public Integer apply(Integer v) throws Throwable {
+                    ts.cancel();
+                    throw new TestException();
+                }
+            })
+            .switchMapMaybe(new Function<Integer, Maybe<Integer>>() {
+                @Override
+                public Maybe<Integer> apply(Integer v) throws Throwable {
+                    return Maybe.just(v).hide();
+                }
+            })
+            .subscribe(ts);
+
+            ts.assertEmpty();
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/mixed/FlowableSwitchMapSingleTest.java

@@ -603,4 +603,34 @@ public void backpressured() {
         .requestMore(1)
         .assertResult(1);
     }
+
+    @Test
+    public void undeliverableUponCancel() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final TestSubscriberEx<Integer> ts = new TestSubscriberEx<Integer>();
+
+            Flowable.just(1)
+            .map(new Function<Integer, Integer>() {
+                @Override
+                public Integer apply(Integer v) throws Throwable {
+                    ts.cancel();
+                    throw new TestException();
+                }
+            })
+            .switchMapSingle(new Function<Integer, Single<Integer>>() {
+                @Override
+                public Single<Integer> apply(Integer v) throws Throwable {
+                    return Single.just(v).hide();
+                }
+            })
+            .subscribe(ts);
+
+            ts.assertEmpty();
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/mixed/ObservableSwitchMapCompletableTest.java

@@ -27,7 +27,7 @@
 import io.reactivex.observers.TestObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subjects.*;
-import io.reactivex.testsupport.TestHelper;
+import io.reactivex.testsupport.*;
 
 public class ObservableSwitchMapCompletableTest {
 
@@ -429,4 +429,34 @@ public void scalarSource() {
 
         to.assertResult();
     }
+
+    @Test
+    public void undeliverableUponCancel() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final TestObserverEx<Integer> to = new TestObserverEx<Integer>();
+
+            Observable.just(1)
+            .map(new Function<Integer, Integer>() {
+                @Override
+                public Integer apply(Integer v) throws Throwable {
+                    to.dispose();
+                    throw new TestException();
+                }
+            })
+            .switchMapCompletable(new Function<Integer, Completable>() {
+                @Override
+                public Completable apply(Integer v) throws Throwable {
+                    return Completable.complete().hide();
+                }
+            })
+            .subscribe(to);
+
+            to.assertEmpty();
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
 }

src/test/java/io/reactivex/internal/operators/mixed/ObservableSwitchMapMaybeTest.java

@@ -686,4 +686,34 @@ public void scalarSource() {
 
         to.assertResult(2);
     }
+
+    @Test
+    public void undeliverableUponCancel() {
+        List<Throwable> errors = TestHelper.trackPluginErrors();
+        try {
+            final TestObserverEx<Integer> to = new TestObserverEx<Integer>();
+
+            Observable.just(1)
+            .map(new Function<Integer, Integer>() {
+                @Override
+                public Integer apply(Integer v) throws Throwable {
+                    to.dispose();
+                    throw new TestException();
+                }
+            })
+            .switchMapMaybe(new Function<Integer, Maybe<Integer>>() {
+                @Override
+                public Maybe<Integer> apply(Integer v) throws Throwable {
+                    return Maybe.just(v).hide();
+                }
+            })
+            .subscribe(to);
+
+            to.assertEmpty();
+
+            TestHelper.assertUndeliverable(errors, 0, TestException.class);
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
 }