From e7842014e9b80304723c488fa1fcffe2b32a228b Mon Sep 17 00:00:00 2001 From: David Karnok Date: Tue, 30 Jul 2019 10:54:54 +0200 Subject: [PATCH] 2.x: Fix mergeWith not canceling other when the main fails (#6599) * 2.x: Fix mergeWith not canceling other when the main fails * Switch to OpenJDK compilation as OracleJDK is not available * Add more time to refCount testing * More time again * Looks like 250ms is still not enough, let's loop --- .travis.yml | 2 +- .../FlowableMergeWithCompletable.java | 2 +- .../flowable/FlowableMergeWithMaybe.java | 2 +- .../flowable/FlowableMergeWithSingle.java | 2 +- .../ObservableMergeWithCompletable.java | 2 +- .../observable/ObservableMergeWithMaybe.java | 2 +- .../observable/ObservableMergeWithSingle.java | 2 +- .../FlowableMergeWithCompletableTest.java | 36 +++++++++++++++++++ .../flowable/FlowableMergeWithMaybeTest.java | 36 +++++++++++++++++++ .../flowable/FlowableMergeWithSingleTest.java | 36 +++++++++++++++++++ .../ObservableMergeWithCompletableTest.java | 36 +++++++++++++++++++ .../ObservableMergeWithMaybeTest.java | 35 ++++++++++++++++++ .../ObservableMergeWithSingleTest.java | 36 +++++++++++++++++++ .../observable/ObservableRefCountAltTest.java | 29 +++++++++------ .../observable/ObservableRefCountTest.java | 16 ++++----- 15 files changed, 249 insertions(+), 25 deletions(-) diff --git a/.travis.yml b/.travis.yml index 83835caeba..9c5c0f6909 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,6 +1,6 @@ language: java jdk: -- oraclejdk8 +- openjdk8 # force upgrade Java8 as per https://github.com/travis-ci/travis-ci/issues/4042 (fixes compilation issue) #addons: diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableMergeWithCompletable.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableMergeWithCompletable.java index 271bd9c50d..c65386bbb1 100644 --- 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 index a32a0c92fc..1787d5fce3 100644 --- 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 index 586bc07c07..486cb73f8c 100644 --- 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 index fa020b6ae4..3b9e649062 100644 --- 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 index 23b2532d9b..e7caad3b21 100644 --- 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 index 20c4d21b5c..7332a29a25 100644 --- 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 index cf5b7917a6..18f5551e4a 100644 --- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableMergeWithCompletableTest.java +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableMergeWithCompletableTest.java @@ -136,4 +136,40 @@ public void run() { ts.assertResult(1); } } + + @Test + public void cancelOtherOnMainError() { + PublishProcessor pp = PublishProcessor.create(); + CompletableSubject cs = CompletableSubject.create(); + + TestSubscriber 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 pp = PublishProcessor.create(); + CompletableSubject cs = CompletableSubject.create(); + + TestSubscriber 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 index c38bf6ae7b..676c9074c3 100644 --- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableMergeWithMaybeTest.java +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableMergeWithMaybeTest.java @@ -401,4 +401,40 @@ public void onNext(Integer t) { ts.assertValueCount(Flowable.bufferSize()); ts.assertComplete(); } + + @Test + public void cancelOtherOnMainError() { + PublishProcessor pp = PublishProcessor.create(); + MaybeSubject ms = MaybeSubject.create(); + + TestSubscriber 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 pp = PublishProcessor.create(); + MaybeSubject ms = MaybeSubject.create(); + + TestSubscriber 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 index 2ab0568a7e..6a182785da 100644 --- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableMergeWithSingleTest.java +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableMergeWithSingleTest.java @@ -397,4 +397,40 @@ public void onNext(Integer t) { ts.assertValueCount(Flowable.bufferSize()); ts.assertComplete(); } + + @Test + public void cancelOtherOnMainError() { + PublishProcessor pp = PublishProcessor.create(); + SingleSubject ss = SingleSubject.create(); + + TestSubscriber 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 pp = PublishProcessor.create(); + SingleSubject ss = SingleSubject.create(); + + TestSubscriber 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 index 872509e164..d9a54c916a 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservableMergeWithCompletableTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableMergeWithCompletableTest.java @@ -135,4 +135,40 @@ protected void subscribeActual(Observer observer) { .test() .assertResult(1); } + + @Test + public void cancelOtherOnMainError() { + PublishSubject ps = PublishSubject.create(); + CompletableSubject cs = CompletableSubject.create(); + + TestObserver 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 ps = PublishSubject.create(); + CompletableSubject cs = CompletableSubject.create(); + + TestObserver 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 index a70e4c2fa8..ee9eb9b576 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservableMergeWithMaybeTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableMergeWithMaybeTest.java @@ -272,4 +272,39 @@ public void onNext(Integer t) { to.assertResult(0, 1, 2, 3, 4); } + @Test + public void cancelOtherOnMainError() { + PublishSubject ps = PublishSubject.create(); + MaybeSubject ms = MaybeSubject.create(); + + TestObserver 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 ps = PublishSubject.create(); + MaybeSubject ms = MaybeSubject.create(); + + TestObserver 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 index 25ce78d486..0d8fb3432b 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservableMergeWithSingleTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableMergeWithSingleTest.java @@ -263,4 +263,40 @@ public void onNext(Integer t) { to.assertResult(0, 1, 2, 3, 4); } + + @Test + public void cancelOtherOnMainError() { + PublishSubject ps = PublishSubject.create(); + SingleSubject ss = SingleSubject.create(); + + TestObserver 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 ps = PublishSubject.create(); + SingleSubject ss = SingleSubject.create(); + + TestObserver 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()); + } } diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountAltTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountAltTest.java index 5fc3aea8c6..05aada6b84 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountAltTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountAltTest.java @@ -630,7 +630,7 @@ protected void subscribeActual(Observer observer) { @Test public void replayNoLeak() throws Exception { System.gc(); - Thread.sleep(100); + Thread.sleep(250); long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -646,7 +646,7 @@ public Object call() throws Exception { source.subscribe(); System.gc(); - Thread.sleep(100); + Thread.sleep(250); long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -657,7 +657,7 @@ public Object call() throws Exception { @Test public void replayNoLeak2() throws Exception { System.gc(); - Thread.sleep(100); + Thread.sleep(250); long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -680,7 +680,7 @@ public Object call() throws Exception { d2 = null; System.gc(); - Thread.sleep(100); + Thread.sleep(250); long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -701,7 +701,7 @@ static final class ExceptionData extends Exception { @Test public void publishNoLeak() throws Exception { System.gc(); - Thread.sleep(100); + Thread.sleep(250); long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -716,10 +716,19 @@ public Object call() throws Exception { source.subscribe(Functions.emptyConsumer(), Functions.emptyConsumer()); - System.gc(); - Thread.sleep(100); + long after = 0L; - long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + for (int i = 0; i < 10; i++) { + System.gc(); + + after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); + + if (start + 20 * 1000 * 1000 > after) { + break; + } + + Thread.sleep(100); + } source = null; assertTrue(String.format("%,3d -> %,3d%n", start, after), start + 20 * 1000 * 1000 > after); @@ -728,7 +737,7 @@ public Object call() throws Exception { @Test public void publishNoLeak2() throws Exception { System.gc(); - Thread.sleep(100); + Thread.sleep(250); long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -751,7 +760,7 @@ public Object call() throws Exception { d2 = null; System.gc(); - Thread.sleep(100); + Thread.sleep(250); long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java index 96be759db2..99a2a79f79 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java @@ -651,7 +651,7 @@ protected void subscribeActual(Observer observer) { @Test public void replayNoLeak() throws Exception { System.gc(); - Thread.sleep(100); + Thread.sleep(250); long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -667,7 +667,7 @@ public Object call() throws Exception { source.subscribe(); System.gc(); - Thread.sleep(100); + Thread.sleep(250); long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -678,7 +678,7 @@ public Object call() throws Exception { @Test public void replayNoLeak2() throws Exception { System.gc(); - Thread.sleep(100); + Thread.sleep(250); long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -701,7 +701,7 @@ public Object call() throws Exception { d2 = null; System.gc(); - Thread.sleep(100); + Thread.sleep(250); long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -722,7 +722,7 @@ static final class ExceptionData extends Exception { @Test public void publishNoLeak() throws Exception { System.gc(); - Thread.sleep(100); + Thread.sleep(250); long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -738,7 +738,7 @@ public Object call() throws Exception { source.subscribe(Functions.emptyConsumer(), Functions.emptyConsumer()); System.gc(); - Thread.sleep(100); + Thread.sleep(250); long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -749,7 +749,7 @@ public Object call() throws Exception { @Test public void publishNoLeak2() throws Exception { System.gc(); - Thread.sleep(100); + Thread.sleep(250); long start = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed(); @@ -772,7 +772,7 @@ public Object call() throws Exception { d2 = null; System.gc(); - Thread.sleep(100); + Thread.sleep(250); long after = ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getUsed();