From 86f23e5a64218b368ed3eed3ba01c87112024f04 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 11 Aug 2026 11:22:00 +0200 Subject: [PATCH 1/2] 4.x: Fix consumer retention in cache() upon subscribe-cancel race --- .../operators/flowable/FlowableCache.java | 7 ++++++ .../operators/observable/ObservableCache.java | 5 +++- .../operators/flowable/FlowableCacheTest.java | 21 ++++++++++++---- .../observable/ObservableCacheTest.java | 25 +++++++++++++++++++ 4 files changed, 52 insertions(+), 6 deletions(-) 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 1d312f76122..01fd1135dc0 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 20eb9251722..c321c79797f 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 060ef77f855..7f291b854ce 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 f65c467ccff..11f8d5eccb3 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!"); + } } From 4351a479e8dab7f8a824e0a60237d7fde146d807 Mon Sep 17 00:00:00 2001 From: akarnokd Date: Tue, 11 Aug 2026 11:35:35 +0200 Subject: [PATCH 2/2] import fix --- .../internal/operators/observable/ObservableFlatMapTest.java | 1 - 1 file changed, 1 deletion(-) 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 1feabb15a59..c8e587cd345 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;