diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCache.java b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCache.java index 1d312f7612..01fd1135dc 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCache.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCache.java @@ -83,6 +83,9 @@ protected void subscribeActual(Subscriber t) { CacheSubscription consumer = new CacheSubscription<>(t, multicaster, head); t.onSubscribe(consumer); multicaster.add(consumer); + if (consumer.isCancelled()) { + multicaster.remove(consumer); + } if (!once.get() && once.compareAndSet(false, true)) { source.subscribe(multicaster); @@ -415,6 +418,10 @@ public void cancel() { parent.remove(this); } } + + public boolean isCancelled() { + return requested.get() == Long.MIN_VALUE; + } } /** diff --git a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCache.java b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCache.java index 20eb925172..c321c79797 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCache.java +++ b/src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCache.java @@ -48,7 +48,7 @@ public final class ObservableCache extends AbstractObservableWithUpstream multicaster; + final Multicaster multicaster; /** * The first node in a singly linked list. Each node has the capacity to hold a specific number of events, and each @@ -79,6 +79,9 @@ protected void subscribeActual(Observer t) { CacheDisposable consumer = new CacheDisposable<>(t, multicaster, head); t.onSubscribe(consumer); multicaster.add(consumer); + if (consumer.isDisposed()) { + multicaster.remove(consumer); + } if (!once.get() && once.compareAndSet(false, true)) { source.subscribe(multicaster); diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCacheTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCacheTest.java index 060ef77f85..7f291b854c 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCacheTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCacheTest.java @@ -16,13 +16,10 @@ import static org.junit.jupiter.api.Assertions.*; import static org.mockito.Mockito.*; -import java.lang.management.ManagementFactory; -import java.lang.management.MemoryMXBean; -import java.lang.management.MemoryUsage; +import java.lang.management.*; import java.util.*; import java.util.concurrent.*; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.*; import org.junit.jupiter.api.Test; @@ -514,4 +511,18 @@ public void valuesAreReclaimable() throws Exception { + " -> " + after.get() / 1024.0 / 1024.0); } } + + @Test + public void noRetentionOnImmediateSelfCancel() { + var source = (FlowableCache)Flowable.never().cache(); + + var consumer = new TestSubscriber(); + consumer.cancel(); + + source.subscribe(consumer); + + consumer.assertNoValues().assertNoErrors().assertNotComplete(); + + assertEquals(0, source.multicaster.get().length, "Retention error!"); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCacheTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCacheTest.java index f65c467ccf..11f8d5eccb 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCacheTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCacheTest.java @@ -370,4 +370,29 @@ public void valuesAreReclaimable() throws Exception { + " -> " + after.get() / 1024.0 / 1024.0); } } + + @Test + public void cancelledUpFrontConnectAnyway() { + final AtomicInteger call = new AtomicInteger(); + Observable.fromCallable(call::incrementAndGet) + .cache() + .test(true) + .assertNoValues(); + + assertEquals(1, call.get()); + } + + @Test + public void noRetentionOnImmediateSelfCancel() { + var source = (ObservableCache)Observable.never().cache(); + + var consumer = new TestObserver(); + consumer.dispose(); + + source.subscribe(consumer); + + consumer.assertNoValues().assertNoErrors().assertNotComplete(); + + assertEquals(0, source.multicaster.get().length, "Retention error!"); + } } diff --git a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableFlatMapTest.java b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableFlatMapTest.java index 1feabb15a5..c8e587cd34 100644 --- a/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableFlatMapTest.java +++ b/src/test/java/io/reactivex/rxjava4/internal/operators/observable/ObservableFlatMapTest.java @@ -32,7 +32,6 @@ import io.reactivex.rxjava4.disposables.Disposable; import io.reactivex.rxjava4.exceptions.*; import io.reactivex.rxjava4.functions.*; -import io.reactivex.rxjava4.internal.functions.Functions; import io.reactivex.rxjava4.observers.TestObserver; import io.reactivex.rxjava4.plugins.RxJavaPlugins; import io.reactivex.rxjava4.schedulers.Schedulers;