3.x: Fix refCount not resetting when termination triggers cross-cancel (#6609)

Author: akarnokd

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();
             }
         }
     }

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();
             }
         }
     }

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<Integer> rp = ReplayProcessor.create();
+        rp.onNext(1);
+        rp.onComplete();
+
+        Flowable<Integer> shared = rp.share();
+
+        shared
+        .buffer(shared.debounce(5, TimeUnit.SECONDS))
+        .test()
+        .assertValueCount(2);
+
+        shared
+        .buffer(shared.debounce(5, TimeUnit.SECONDS))
+        .test()
+        .assertValueCount(2);
+    }}

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<Integer> rs = ReplaySubject.create();
+        rs.onNext(1);
+        rs.onComplete();
+
+        Observable<Integer> shared = rs.share();
+
+        shared
+        .buffer(shared.debounce(5, TimeUnit.SECONDS))
+        .test()
+        .assertValueCount(2);
+
+        shared
+        .buffer(shared.debounce(5, TimeUnit.SECONDS))
+        .test()
+        .assertValueCount(2);
+    }
 }