3.x: Fix mergeWith not cancelling the other source if the main errors (#6598)

* 3.x: Fix mergeWith not cancelling the other source if the main errors

* Fix typo in the test name

Author: akarnokd

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

@@ -86,7 +86,7 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable ex) {
-            SubscriptionHelper.cancel(mainSubscription);
+            DisposableHelper.dispose(otherObserver);
             HalfSerializer.onError(downstream, ex, this, error);
         }
 

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

@@ -143,7 +143,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable ex) {
             if (error.addThrowable(ex)) {
-                SubscriptionHelper.cancel(mainSubscription);
+                DisposableHelper.dispose(otherObserver);
                 drain();
             } else {
                 RxJavaPlugins.onError(ex);

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

@@ -143,7 +143,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable ex) {
             if (error.addThrowable(ex)) {
-                SubscriptionHelper.cancel(mainSubscription);
+                DisposableHelper.dispose(otherObserver);
                 drain();
             } else {
                 RxJavaPlugins.onError(ex);

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

@@ -80,7 +80,7 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable ex) {
-            DisposableHelper.dispose(mainDisposable);
+            DisposableHelper.dispose(otherObserver);
             HalfSerializer.onError(downstream, ex, this, error);
         }
 

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

@@ -106,7 +106,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable ex) {
             if (error.addThrowable(ex)) {
-                DisposableHelper.dispose(mainDisposable);
+                DisposableHelper.dispose(otherObserver);
                 drain();
             } else {
                 RxJavaPlugins.onError(ex);

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

@@ -106,7 +106,7 @@ public void onNext(T t) {
         @Override
         public void onError(Throwable ex) {
             if (error.addThrowable(ex)) {
-                DisposableHelper.dispose(mainDisposable);
+                DisposableHelper.dispose(otherObserver);
                 drain();
             } else {
                 RxJavaPlugins.onError(ex);

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

@@ -13,10 +13,10 @@
 
 package io.reactivex.internal.operators.flowable;
 
-import org.junit.Test;
-
 import static org.junit.Assert.*;
 
+import org.junit.Test;
+
 import io.reactivex.*;
 import io.reactivex.exceptions.TestException;
 import io.reactivex.functions.Action;
@@ -137,4 +137,40 @@ public void run() {
             ts.assertResult(1);
         }
     }
+
+    @Test
+    public void cancelOtherOnMainError() {
+        PublishProcessor<Integer> pp = PublishProcessor.create();
+        CompletableSubject cs = CompletableSubject.create();
+
+        TestSubscriber<Integer> ts = pp.mergeWith(cs).test();
+
+        assertTrue(pp.hasSubscribers());
+        assertTrue(cs.hasObservers());
+
+        pp.onError(new TestException());
+
+        ts.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", pp.hasSubscribers());
+        assertFalse("other has observers", cs.hasObservers());
+    }
+
+    @Test
+    public void cancelMainOnOtherError() {
+        PublishProcessor<Integer> pp = PublishProcessor.create();
+        CompletableSubject cs = CompletableSubject.create();
+
+        TestSubscriber<Integer> ts = pp.mergeWith(cs).test();
+
+        assertTrue(pp.hasSubscribers());
+        assertTrue(cs.hasObservers());
+
+        cs.onError(new TestException());
+
+        ts.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", pp.hasSubscribers());
+        assertFalse("other has observers", cs.hasObservers());
+    }
 }

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

@@ -28,7 +28,7 @@
 import io.reactivex.internal.subscriptions.BooleanSubscription;
 import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.processors.PublishProcessor;
-import io.reactivex.subjects.MaybeSubject;
+import io.reactivex.subjects.*;
 import io.reactivex.subscribers.TestSubscriber;
 import io.reactivex.testsupport.TestHelper;
 
@@ -402,4 +402,40 @@ public void onNext(Integer t) {
         ts.assertValueCount(Flowable.bufferSize());
         ts.assertComplete();
     }
+
+    @Test
+    public void cancelOtherOnMainError() {
+        PublishProcessor<Integer> pp = PublishProcessor.create();
+        MaybeSubject<Integer> ms = MaybeSubject.create();
+
+        TestSubscriber<Integer> ts = pp.mergeWith(ms).test();
+
+        assertTrue(pp.hasSubscribers());
+        assertTrue(ms.hasObservers());
+
+        pp.onError(new TestException());
+
+        ts.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", pp.hasSubscribers());
+        assertFalse("other has observers", ms.hasObservers());
+    }
+
+    @Test
+    public void cancelMainOnOtherError() {
+        PublishProcessor<Integer> pp = PublishProcessor.create();
+        MaybeSubject<Integer> ms = MaybeSubject.create();
+
+        TestSubscriber<Integer> ts = pp.mergeWith(ms).test();
+
+        assertTrue(pp.hasSubscribers());
+        assertTrue(ms.hasObservers());
+
+        ms.onError(new TestException());
+
+        ts.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", pp.hasSubscribers());
+        assertFalse("other has observers", ms.hasObservers());
+    }
 }

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

@@ -398,4 +398,40 @@ public void onNext(Integer t) {
         ts.assertValueCount(Flowable.bufferSize());
         ts.assertComplete();
     }
+
+    @Test
+    public void cancelOtherOnMainError() {
+        PublishProcessor<Integer> pp = PublishProcessor.create();
+        SingleSubject<Integer> ss = SingleSubject.create();
+
+        TestSubscriber<Integer> ts = pp.mergeWith(ss).test();
+
+        assertTrue(pp.hasSubscribers());
+        assertTrue(ss.hasObservers());
+
+        pp.onError(new TestException());
+
+        ts.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", pp.hasSubscribers());
+        assertFalse("other has observers", ss.hasObservers());
+    }
+
+    @Test
+    public void cancelMainOnOtherError() {
+        PublishProcessor<Integer> pp = PublishProcessor.create();
+        SingleSubject<Integer> ss = SingleSubject.create();
+
+        TestSubscriber<Integer> ts = pp.mergeWith(ss).test();
+
+        assertTrue(pp.hasSubscribers());
+        assertTrue(ss.hasObservers());
+
+        ss.onError(new TestException());
+
+        ts.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", pp.hasSubscribers());
+        assertFalse("other has observers", ss.hasObservers());
+    }
 }

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

@@ -136,4 +136,40 @@ protected void subscribeActual(Observer<? super Integer> observer) {
         .test()
         .assertResult(1);
     }
+
+    @Test
+    public void cancelOtherOnMainError() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+        CompletableSubject cs = CompletableSubject.create();
+
+        TestObserver<Integer> to = ps.mergeWith(cs).test();
+
+        assertTrue(ps.hasObservers());
+        assertTrue(cs.hasObservers());
+
+        ps.onError(new TestException());
+
+        to.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", ps.hasObservers());
+        assertFalse("other has observers", cs.hasObservers());
+    }
+
+    @Test
+    public void cancelMainOnOtherError() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+        CompletableSubject cs = CompletableSubject.create();
+
+        TestObserver<Integer> to = ps.mergeWith(cs).test();
+
+        assertTrue(ps.hasObservers());
+        assertTrue(cs.hasObservers());
+
+        cs.onError(new TestException());
+
+        to.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", ps.hasObservers());
+        assertFalse("other has observers", cs.hasObservers());
+    }
 }