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

akarnokd · web-flow · commit e00c4993cf73 · 2019-07-30T09:27:03.000+02:00
* 3.x: Fix mergeWith not cancelling the other source if the main errors

* Fix typo in the test name
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableMergeWithCompletable.java b/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);
         }
 
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableMergeWithMaybe.java b/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);
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableMergeWithSingle.java b/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);
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableMergeWithCompletable.java b/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);
         }
 
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableMergeWithMaybe.java b/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);
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableMergeWithSingle.java b/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);
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableMergeWithCompletableTest.java b/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());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableMergeWithMaybeTest.java b/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());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableMergeWithSingleTest.java b/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());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableMergeWithCompletableTest.java b/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());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableMergeWithMaybeTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableMergeWithMaybeTest.java
@@ -273,4 +273,39 @@ public void onNext(Integer t) {
         to.assertResult(0, 1, 2, 3, 4);
     }
 
+    @Test
+    public void cancelOtherOnMainError() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+        MaybeSubject<Integer> ms = MaybeSubject.create();
+
+        TestObserver<Integer> to = ps.mergeWith(ms).test();
+
+        assertTrue(ps.hasObservers());
+        assertTrue(ms.hasObservers());
+
+        ps.onError(new TestException());
+
+        to.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", ps.hasObservers());
+        assertFalse("other has observers", ms.hasObservers());
+    }
+
+    @Test
+    public void cancelMainOnOtherError() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+        MaybeSubject<Integer> ms = MaybeSubject.create();
+
+        TestObserver<Integer> to = ps.mergeWith(ms).test();
+
+        assertTrue(ps.hasObservers());
+        assertTrue(ms.hasObservers());
+
+        ms.onError(new TestException());
+
+        to.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", ps.hasObservers());
+        assertFalse("other has observers", ms.hasObservers());
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableMergeWithSingleTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableMergeWithSingleTest.java
@@ -264,4 +264,40 @@ public void onNext(Integer t) {
 
         to.assertResult(0, 1, 2, 3, 4);
     }
+
+    @Test
+    public void cancelOtherOnMainError() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+        SingleSubject<Integer> ss = SingleSubject.create();
+
+        TestObserver<Integer> to = ps.mergeWith(ss).test();
+
+        assertTrue(ps.hasObservers());
+        assertTrue(ss.hasObservers());
+
+        ps.onError(new TestException());
+
+        to.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", ps.hasObservers());
+        assertFalse("other has observers", ss.hasObservers());
+    }
+
+    @Test
+    public void cancelMainOnOtherError() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+        SingleSubject<Integer> ss = SingleSubject.create();
+
+        TestObserver<Integer> to = ps.mergeWith(ss).test();
+
+        assertTrue(ps.hasObservers());
+        assertTrue(ss.hasObservers());
+
+        ss.onError(new TestException());
+
+        to.assertFailure(TestException.class);
+
+        assertFalse("main has observers!", ps.hasObservers());
+        assertFalse("other has observers", ss.hasObservers());
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -86,7 +86,7 @@ public void onNext(T t) {`
`86`	`86`
`87`	`87`	`@Override`
`88`	`88`	`public void onError(Throwable ex) {`
`89`		`- SubscriptionHelper.cancel(mainSubscription);`
	`89`	`+ DisposableHelper.dispose(otherObserver);`
`90`	`90`	`HalfSerializer.onError(downstream, ex, this, error);`
`91`	`91`	`}`
`92`	`92`
Original file line number	Diff line number	Diff line change
`@@ -80,7 +80,7 @@ public void onNext(T t) {`
`80`	`80`
`81`	`81`	`@Override`
`82`	`82`	`public void onError(Throwable ex) {`
`83`		`- DisposableHelper.dispose(mainDisposable);`
	`83`	`+ DisposableHelper.dispose(otherObserver);`
`84`	`84`	`HalfSerializer.onError(downstream, ex, this, error);`
`85`	`85`	`}`
`86`	`86`