ObserveOn Cleanup

- make code easier to understand with some refactoring of the pollQueue loop
- add a unit test to ensure correct behavior with a hot Observable source

Author: benjchristensen

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

@@ -26,7 +26,6 @@
 import rx.exceptions.MissingBackpressureException;
 import rx.functions.Action0;
 import rx.internal.util.RxRingBuffer;
-import rx.internal.util.SynchronizedSubscription;
 import rx.schedulers.ImmediateScheduler;
 import rx.schedulers.TrampolineScheduler;
 
@@ -100,9 +99,8 @@ public void request(long n) {
             add(scheduledUnsubscribe);
             child.add(recursiveScheduler);
             child.add(this);
-            
         }
-        
+
         @Override
         public void onStart() {
             // signal that this is an async operator capable of receiving this many
@@ -160,9 +158,16 @@ public void call() {
             }
         }
 
+        // only execute this from schedule()
         private void pollQueue() {
             int emitted = 0;
-            while (true) {
+            do {
+                /*
+                 * Set to 1 otherwise it could have grown very large while in the last poll loop
+                 * and then we can end up looping all those times again here before exiting even once we've drained
+                 */
+                COUNTER_UPDATER.set(this, 1);
+
                 while (!scheduledUnsubscribe.isUnsubscribed()) {
                     if (REQUESTED.getAndDecrement(this) != 0) {
                         Object o = queue.poll();
@@ -174,15 +179,18 @@ private void pollQueue() {
                             if (failure) {
                                 // completed so we will skip onNext if they exist and only emit terminal events
                                 if (on.isError(o)) {
-                                    System.out.println("Error: " + o);
                                     // only emit error
                                     on.accept(child, o);
+                                    // TODO this could hit the requested limit again ... and is skipping values
+                                    // so the request count is broken ... it needs to purge the queue
+                                    // or modify the requested amount so it will loop through everything
                                 }
                             } else {
                                 if (!on.accept(child, o)) {
                                     // non-terminal event so let's increment count
                                     emitted++;
                                 }
+
                             }
                         }
                     } else {
@@ -191,22 +199,11 @@ private void pollQueue() {
                         break;
                     }
                 }
-                long c = COUNTER_UPDATER.decrementAndGet(this);
-                if (c <= 0) {
-                    // request the number of items that we emitted in this poll loop
-                    if (emitted > 0) {
-                        request(emitted);
-                    }
-                    break;
-                } else {
-                    /*
-                     * Set down to 1 and then iterate again.
-                     * we lower it to 1 otherwise it could have grown very large while in the last poll loop
-                     * and then we can end up looping all those times again here before existing even once we've drained
-                     */
-                    COUNTER_UPDATER.set(this, 1);
-                    // we now loop again, and if anything tries scheduling again after this it will increment and cause us to loop again after
-                }
+            } while (COUNTER_UPDATER.decrementAndGet(this) > 0);
+
+            // request the number of items that we emitted in this poll loop
+            if (emitted > 0) {
+                request(emitted);
             }
         }
     }

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

@@ -391,7 +391,7 @@ public void testDelayedErrorDeliveryWhenSafeSubscriberUnsubscribes() {
         inOrder.verify(o, never()).onNext(anyInt());
         inOrder.verify(o, never()).onCompleted();
     }
-    
+
     @Test
     public void testAfterUnsubscribeCalledThenObserverOnNextNeverCalled() {
         final TestScheduler testScheduler = new TestScheduler();
@@ -647,6 +647,33 @@ public void onNext(Long t) {
         assertTrue(ts.getOnNextEvents().size() == ts.getOnNextEvents().get(ts.getOnNextEvents().size() - 1) + 1);
         // we should emit the error without emitting the full buffer size
         assertTrue(ts.getOnNextEvents().size() < RxRingBuffer.SIZE);
+    }
+
+    /**
+     * Make sure we get a MissingBackpressureException propagated through when we have a fast temporal (hot) producer.
+     */
+    @Test
+    public void testHotOperatorBackpressure() {
+        TestSubscriber<String> ts = new TestSubscriber<String>();
+        Observable.timer(0, 1, TimeUnit.MILLISECONDS)
+                .observeOn(Schedulers.computation())
+                .map(new Func1<Long, String>() {
 
+                    @Override
+                    public String call(Long t1) {
+                        System.out.println(t1);
+                        try {
+                            Thread.sleep(100);
+                        } catch (InterruptedException e) {
+                        }
+                        return t1 + " slow value";
+                    }
+
+                }).subscribe(ts);
+
+        ts.awaitTerminalEvent();
+        System.out.println("Errors: " + ts.getOnErrorEvents());
+        assertEquals(MissingBackpressureException.class, ts.getOnErrorEvents().get(0).getClass());
     }
+
 }