Merge pull request #2788 from zsxwing/publish-no-subscriber

Fix the bug that 'publish' will cache items when no subscriber

Author: benjchristensen

src/main/java/rx/internal/operators/OperatorPublish.java

@@ -213,6 +213,10 @@ public synchronized boolean canEmitWithDecrement() {
             return false;
         }
 
+        public synchronized boolean hasNoSubscriber() {
+            return subscribers.length == 0;
+        }
+
         public synchronized void incrementOutstandingAfterFailedEmit() {
             outstandingRequests++;
         }
@@ -308,11 +312,13 @@ public void emit(Object t) throws MissingBackpressureException {
         }
 
         private void requestMoreAfterEmission(int emitted) {
-            OriginSubscriber<T> origin = state.getOrigin();
-            if (emitted > 0 && origin != null) {
-                long r = origin.originOutstanding.addAndGet(-emitted);
-                if (r <= origin.THRESHOLD) {
-                    origin.requestMore(RxRingBuffer.SIZE - origin.THRESHOLD);
+            if (emitted > 0) {
+                OriginSubscriber<T> origin = state.getOrigin();
+                if (origin != null) {
+                    long r = origin.originOutstanding.addAndGet(-emitted);
+                    if (r <= origin.THRESHOLD) {
+                        origin.requestMore(RxRingBuffer.SIZE - origin.THRESHOLD);
+                    }
                 }
             }
         }
@@ -336,8 +342,18 @@ public void drainQueue(OriginSubscriber<T> originSubscriber) {
                      * If we want to batch this then we need to account for new subscribers arriving with a lower request count
                      * concurrently while iterating the batch ... or accept that they won't
                      */
-                    
                     while (true) {
+                        if (localState.hasNoSubscriber()) {
+                            // Drop items due to no subscriber
+                            if (localBuffer.poll() == null) {
+                                // Exit due to no more item
+                                break;
+                            } else {
+                                // Keep dropping cached items.
+                                continue;
+                            }
+                        }
+
                         boolean shouldEmit = localState.canEmitWithDecrement();
                         if (!shouldEmit) {
                             break;

src/test/java/rx/internal/operators/OperatorPublishTest.java

@@ -30,6 +30,7 @@
 import rx.observables.ConnectableObservable;
 import rx.observers.TestSubscriber;
 import rx.schedulers.Schedulers;
+import rx.schedulers.TestScheduler;
 
 public class OperatorPublishTest {
 
@@ -242,4 +243,20 @@ public void call() {
 
         assertEquals(8, sourceEmission.get());
     }
+
+    @Test
+    public void testConnectWithNoSubscriber() {
+        TestScheduler scheduler = new TestScheduler();
+        ConnectableObservable<Long> co = Observable.timer(10, 10, TimeUnit.MILLISECONDS, scheduler).take(3).publish();
+        co.connect();
+        // Emit 0
+        scheduler.advanceTimeBy(15, TimeUnit.MILLISECONDS);
+        TestSubscriber subscriber = new TestSubscriber<Long>();
+        co.subscribe(subscriber);
+        // Emit 1 and 2
+        scheduler.advanceTimeBy(50, TimeUnit.MILLISECONDS);
+        subscriber.assertReceivedOnNext(Arrays.asList(1L, 2L));
+        subscriber.assertNoErrors();
+        subscriber.assertTerminalEvent();
+    }
 }