2.x: Fix publish().refCount() hang due to race (#6505)

* 2.x: Fix publish().refCount() hang due to race

* Add more time to GC when detecting leaks.

* Fix subscriber swap mistake in the Alt implementation

Author: akarnokd

src/main/java/io/reactivex/flowables/ConnectableFlowable.java

@@ -68,6 +68,24 @@ public final Disposable connect() {
         return cc.disposable;
     }
 
+    /**
+     * Apply a workaround for a race condition with the regular publish().refCount()
+     * so that racing subscribers and refCount won't hang.
+     * 
+     * @return the ConnectableFlowable to work with
+     * @since 2.2.10
+     */
+    private ConnectableFlowable<T> onRefCount() {
+        if (this instanceof FlowablePublishClassic) {
+            @SuppressWarnings("unchecked")
+            FlowablePublishClassic<T> fp = (FlowablePublishClassic<T>) this;
+            return RxJavaPlugins.onAssembly(
+                    new FlowablePublishAlt<T>(fp.publishSource(), fp.publishBufferSize())
+                    );
+        }
+        return this;
+    }
+
     /**
      * Returns a {@code Flowable} that stays connected to this {@code ConnectableFlowable} as long as there
      * is at least one subscription to this {@code ConnectableFlowable}.
@@ -89,7 +107,7 @@ public final Disposable connect() {
     @SchedulerSupport(SchedulerSupport.NONE)
     @BackpressureSupport(BackpressureKind.PASS_THROUGH)
     public Flowable<T> refCount() {
-        return RxJavaPlugins.onAssembly(new FlowableRefCount<T>(this));
+        return RxJavaPlugins.onAssembly(new FlowableRefCount<T>(onRefCount()));
     }
 
     /**
@@ -216,7 +234,7 @@ public final Flowable<T> refCount(int subscriberCount, long timeout, TimeUnit un
         ObjectHelper.verifyPositive(subscriberCount, "subscriberCount");
         ObjectHelper.requireNonNull(unit, "unit is null");
         ObjectHelper.requireNonNull(scheduler, "scheduler is null");
-        return RxJavaPlugins.onAssembly(new FlowableRefCount<T>(this, subscriberCount, timeout, unit, scheduler));
+        return RxJavaPlugins.onAssembly(new FlowableRefCount<T>(onRefCount(), subscriberCount, timeout, unit, scheduler));
     }
 
     /**

src/main/java/io/reactivex/internal/operators/flowable/FlowablePublish.java

@@ -33,7 +33,8 @@
  * manner.
  * @param <T> the value type
  */
-public final class FlowablePublish<T> extends ConnectableFlowable<T> implements HasUpstreamPublisher<T> {
+public final class FlowablePublish<T> extends ConnectableFlowable<T>
+implements HasUpstreamPublisher<T>, FlowablePublishClassic<T> {
     /**
      * Indicates this child has been cancelled: the state is swapped in atomically and
      * will prevent the dispatch() to emit (too many) values to a terminated child subscriber.
@@ -77,6 +78,19 @@ public Publisher<T> source() {
         return source;
     }
 
+    /**
+     * @return The internal buffer size of this FloawblePublish operator.
+     */
+    @Override
+    public int publishBufferSize() {
+        return bufferSize;
+    }
+
+    @Override
+    public Publisher<T> publishSource() {
+        return source;
+    }
+
     @Override
     protected void subscribeActual(Subscriber<? super T> s) {
         onSubscribe.subscribe(s);