2.x: Fix Observable.switchMap main onError not disposing the current inner source (#5833)

* 2.x: Fix Obs.switchMap main onError not disposing the current inner src

* Fix error-error race test

Author: akarnokd

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

@@ -131,15 +131,15 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable t) {
-            if (done || !errors.addThrowable(t)) {
+            if (!done && errors.addThrowable(t)) {
                 if (!delayErrors) {
                     disposeInner();
                 }
+                done = true;
+                drain();
+            } else {
                 RxJavaPlugins.onError(t);
-                return;
             }
-            done = true;
-            drain();
         }
 
         @Override

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

@@ -1153,4 +1153,25 @@ public Object apply(Integer v) throws Exception {
         .test()
         .assertFailure(TestException.class);
     }
+
+    @Test
+    public void innerCancelledOnMainError() {
+        final PublishProcessor<Integer> main = PublishProcessor.create();
+        final PublishProcessor<Integer> inner = PublishProcessor.create();
+
+        TestSubscriber<Integer> to = main.switchMap(Functions.justFunction(inner))
+        .test();
+
+        assertTrue(main.hasSubscribers());
+
+        main.onNext(1);
+
+        assertTrue(inner.hasSubscribers());
+
+        main.onError(new TestException());
+
+        assertFalse(inner.hasSubscribers());
+
+        to.assertFailure(TestException.class);
+    }
 }

src/test/java/io/reactivex/internal/operators/observable/ObservableSwitchTest.java

@@ -18,7 +18,7 @@
 
 import java.util.List;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.*;
 
 import org.junit.*;
 import org.mockito.InOrder;
@@ -768,7 +768,7 @@ public void run() {
 
     @Test
     public void outerInnerErrorRace() {
-        for (int i = 0; i < 500; i++) {
+        for (int i = 0; i < 5000; i++) {
             List<Throwable> errors = TestHelper.trackPluginErrors();
             try {
 
@@ -786,6 +786,8 @@ public ObservableSource<Integer> apply(Integer v) throws Exception {
                 })
                 .test();
 
+                ps1.onNext(1);
+
                 final TestException ex1 = new TestException();
 
                 Runnable r1 = new Runnable() {
@@ -807,7 +809,7 @@ public void run() {
                 TestHelper.race(r1, r2);
 
                 for (Throwable e : errors) {
-                    assertTrue(e.toString(), e instanceof TestException);
+                    assertTrue(e.getCause().toString(), e.getCause() instanceof TestException);
                 }
             } finally {
                 RxJavaPlugins.reset();
@@ -963,4 +965,93 @@ public void onNext(Integer t) {
 
         to.assertFailure(TestException.class, 1);
     }
+
+    @Test
+    public void innerDisposedOnMainError() {
+        final PublishSubject<Integer> main = PublishSubject.create();
+        final PublishSubject<Integer> inner = PublishSubject.create();
+
+        TestObserver<Integer> to = main.switchMap(Functions.justFunction(inner))
+        .test();
+
+        assertTrue(main.hasObservers());
+
+        main.onNext(1);
+
+        assertTrue(inner.hasObservers());
+
+        main.onError(new TestException());
+
+        assertFalse(inner.hasObservers());
+
+        to.assertFailure(TestException.class);
+    }
+
+    @Test
+    public void outerInnerErrorRaceIgnoreDispose() {
+        for (int i = 0; i < 5000; i++) {
+            List<Throwable> errors = TestHelper.trackPluginErrors();
+            try {
+
+                final AtomicReference<Observer<? super Integer>> obs1 = new AtomicReference<Observer<? super Integer>>();
+                final Observable<Integer> ps1 = new Observable<Integer>() {
+                    @Override
+                    protected void subscribeActual(
+                            Observer<? super Integer> observer) {
+                        obs1.set(observer);
+                    }
+                };
+                final AtomicReference<Observer<? super Integer>> obs2 = new AtomicReference<Observer<? super Integer>>();
+                final Observable<Integer> ps2 = new Observable<Integer>() {
+                    @Override
+                    protected void subscribeActual(
+                            Observer<? super Integer> observer) {
+                        obs2.set(observer);
+                    }
+                };
+
+                ps1.switchMap(new Function<Integer, ObservableSource<Integer>>() {
+                    @Override
+                    public ObservableSource<Integer> apply(Integer v) throws Exception {
+                        if (v == 1) {
+                            return ps2;
+                        }
+                        return Observable.never();
+                    }
+                })
+                .test();
+
+                obs1.get().onSubscribe(Disposables.empty());
+                obs1.get().onNext(1);
+
+                obs2.get().onSubscribe(Disposables.empty());
+
+                final TestException ex1 = new TestException();
+
+                Runnable r1 = new Runnable() {
+                    @Override
+                    public void run() {
+                        obs1.get().onError(ex1);
+                    }
+                };
+
+                final TestException ex2 = new TestException();
+
+                Runnable r2 = new Runnable() {
+                    @Override
+                    public void run() {
+                        obs2.get().onError(ex2);
+                    }
+                };
+
+                TestHelper.race(r1, r2);
+
+                for (Throwable e : errors) {
+                    assertTrue(e.toString(), e.getCause() instanceof TestException);
+                }
+            } finally {
+                RxJavaPlugins.reset();
+            }
+        }
+    }
 }