From cb49e407924a5a635d5743f9d66be02ad1546062 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 30 Jul 2019 09:49:40 +0200 Subject: [PATCH 1/5] 2.x: Fix mergeWith not canceling other when the main fails --- .../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 +++++++++++++++++++ 12 files changed, 221 insertions(+), 6 deletions(-) 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()); + } } From 3322be5735cf7cee77c2e31b8b01a73fa6dc2b6f Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 30 Jul 2019 09:55:01 +0200 Subject: [PATCH 2/5] Switch to OpenJDK compilation as OracleJDK is not available --- .travis.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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: From 19639de1e673edc10f3f878897d313b6fa5c3e81 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 30 Jul 2019 10:10:44 +0200 Subject: [PATCH 3/5] Add more time to refCount testing --- .../operators/observable/ObservableRefCountAltTest.java | 8 ++++---- .../operators/observable/ObservableRefCountTest.java | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) 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..a213dd4a58 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountAltTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountAltTest.java @@ -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(); @@ -717,7 +717,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(); @@ -728,7 +728,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 +751,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..a1974e09d2 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java @@ -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(); From 16587ccf1fa4dd784c332112f87f5abb1c895e1b Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 30 Jul 2019 10:21:36 +0200 Subject: [PATCH 4/5] More time again --- .../operators/observable/ObservableRefCountAltTest.java | 8 ++++---- .../operators/observable/ObservableRefCountTest.java | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) 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 a213dd4a58..4d1007883c 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(); 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 a1974e09d2..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(); From b1c75facd0f60e13cdbefc7c3bbd9cb484c5e786 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 30 Jul 2019 10:29:58 +0200 Subject: [PATCH 5/5] Looks like 250ms is still not enough, let's loop --- .../observable/ObservableRefCountAltTest.java | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) 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 4d1007883c..05aada6b84 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountAltTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountAltTest.java @@ -716,10 +716,19 @@ public Object call() throws Exception { source.subscribe(Functions.emptyConsumer(), Functions.emptyConsumer()); - System.gc(); - Thread.sleep(250); + 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);