diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRefCount.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRefCount.java index 27557ebc79..414dfab8bb 100644 --- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableRefCount.java +++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableRefCount.java @@ -115,14 +115,15 @@ void cancel(RefConnection rc) { void terminated(RefConnection rc) { synchronized (this) { - if (connection != null && connection == rc) { - connection = null; + if (connection == rc) { if (rc.timer != null) { rc.timer.dispose(); + rc.timer = null; + } + if (--rc.subscriberCount == 0) { + connection = null; + source.reset(); } - } - if (--rc.subscriberCount == 0) { - source.reset(); } } } diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableRefCount.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableRefCount.java index c238403c20..57d99325e6 100644 --- a/src/main/java/io/reactivex/internal/operators/observable/ObservableRefCount.java +++ b/src/main/java/io/reactivex/internal/operators/observable/ObservableRefCount.java @@ -112,14 +112,15 @@ void cancel(RefConnection rc) { void terminated(RefConnection rc) { synchronized (this) { - if (connection != null && connection == rc) { - connection = null; + if (connection == rc) { if (rc.timer != null) { rc.timer.dispose(); + rc.timer = null; + } + if (--rc.subscriberCount == 0) { + connection = null; + source.reset(); } - } - if (--rc.subscriberCount == 0) { - source.reset(); } } } diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountTest.java index 5e12bd5214..fd0259b2dd 100644 --- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountTest.java +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableRefCountTest.java @@ -14,6 +14,7 @@ package io.reactivex.internal.operators.flowable; import static org.junit.Assert.*; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.*; import java.io.IOException; @@ -1454,4 +1455,22 @@ public void publishRefCountShallBeThreadSafe() { .assertComplete(); } } -} + + @Test + public void upstreamTerminationTriggersAnotherCancel() throws Exception { + ReplayProcessor rp = ReplayProcessor.create(); + rp.onNext(1); + rp.onComplete(); + + Flowable shared = rp.share(); + + shared + .buffer(shared.debounce(5, TimeUnit.SECONDS)) + .test() + .assertValueCount(2); + + shared + .buffer(shared.debounce(5, TimeUnit.SECONDS)) + .test() + .assertValueCount(2); + }} 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 c3db4fe0e8..cc7b04e6e1 100644 --- a/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java +++ b/src/test/java/io/reactivex/internal/operators/observable/ObservableRefCountTest.java @@ -14,6 +14,7 @@ package io.reactivex.internal.operators.observable; import static org.junit.Assert.*; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.*; import java.io.IOException; @@ -22,7 +23,7 @@ import java.util.concurrent.*; import java.util.concurrent.atomic.*; -import org.junit.Test; +import org.junit.*; import org.mockito.InOrder; import io.reactivex.*; @@ -1267,8 +1268,6 @@ public void cancelTerminateStateExclusion() { .publish() .refCount(); - o.cancel(null); - o.cancel(new RefConnection(o)); RefConnection rc = new RefConnection(o); @@ -1412,4 +1411,23 @@ public void publishRefCountShallBeThreadSafe() { .assertComplete(); } } + + @Test + public void upstreamTerminationTriggersAnotherCancel() throws Exception { + ReplaySubject rs = ReplaySubject.create(); + rs.onNext(1); + rs.onComplete(); + + Observable shared = rs.share(); + + shared + .buffer(shared.debounce(5, TimeUnit.SECONDS)) + .test() + .assertValueCount(2); + + shared + .buffer(shared.debounce(5, TimeUnit.SECONDS)) + .test() + .assertValueCount(2); + } }