4.x: Fix consumer retention in cache() upon subscribe-cancel race

src/main/java/io/reactivex/rxjava4/internal/operators/flowable/FlowableCache.java

@@ -83,6 +83,9 @@ protected void subscribeActual(Subscriber<? super T> t) {
         CacheSubscription<T> 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;
+        }
     }
 
     /**

src/main/java/io/reactivex/rxjava4/internal/operators/observable/ObservableCache.java

@@ -48,7 +48,7 @@ public final class ObservableCache<T> extends AbstractObservableWithUpstream<T,
     /**
      * Responsible caching events from the source and multicasting them to each downstream.
      */
-    final Multicaster<T>  multicaster;
+    final Multicaster<T> 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<? super T> t) {
         CacheDisposable<T> 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);

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<Integer>)Flowable.<Integer>never().cache();
+
+        var consumer = new TestSubscriber<Integer>();
+        consumer.cancel();
+
+        source.subscribe(consumer);
+
+        consumer.assertNoValues().assertNoErrors().assertNotComplete();
+
+        assertEquals(0, source.multicaster.get().length, "Retention error!");
+    }
 }

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<Integer>)Observable.<Integer>never().cache();
+
+        var consumer = new TestObserver<Integer>();
+        consumer.dispose();
+
+        source.subscribe(consumer);
+
+        consumer.assertNoValues().assertNoErrors().assertNotComplete();
+
+        assertEquals(0, source.multicaster.get().length, "Retention error!");
+    }
 }

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;