ObserveOn Error Propagation

- errors were not being propagated if backpressure was exerted (requested==0)
- terminal events do not need to be requested so it now ignores the requested count to emit onError

Author: benjchristensen

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

@@ -96,7 +96,6 @@ public void request(long n) {
                 }
 
             });
-            add(scheduledUnsubscribe);
             child.add(recursiveScheduler);
             child.add(this);
         }
@@ -169,34 +168,34 @@ private void pollQueue() {
                 COUNTER_UPDATER.set(this, 1);
 
                 while (!scheduledUnsubscribe.isUnsubscribed()) {
-                    if (REQUESTED.getAndDecrement(this) != 0) {
+                    if (failure) {
+                        // special handling to short-circuit an error propagation
                         Object o = queue.poll();
-                        if (o == null) {
-                            // nothing in queue
-                            REQUESTED.incrementAndGet(this);
-                            break;
-                        } else {
-                            if (failure) {
-                                // completed so we will skip onNext if they exist and only emit terminal events
-                                if (on.isError(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
-                                }
+                        // completed so we will skip onNext if they exist and only emit terminal events
+                        if (on.isError(o)) {
+                            // only emit error
+                            on.accept(child, o);
+                            // we have emitted a terminal event so return (exit the loop we're in)
+                            return;
+                        }
+                    } else {
+                        if (REQUESTED.getAndDecrement(this) != 0) {
+                            Object o = queue.poll();
+                            if (o == null) {
+                                // nothing in queue
+                                REQUESTED.incrementAndGet(this);
+                                break;
                             } else {
                                 if (!on.accept(child, o)) {
                                     // non-terminal event so let's increment count
                                     emitted++;
                                 }
-
                             }
+                        } else {
+                            // we hit the end ... so increment back to 0 again
+                            REQUESTED.incrementAndGet(this);
+                            break;
                         }
-                    } else {
-                        // we hit the end ... so increment back to 0 again
-                        REQUESTED.incrementAndGet(this);
-                        break;
                     }
                 }
             } while (COUNTER_UPDATER.decrementAndGet(this) > 0);

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

@@ -34,9 +34,11 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.junit.Ignore;
 import org.junit.Test;
 import org.mockito.InOrder;
 
+import rx.Notification;
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.Observer;
@@ -48,6 +50,7 @@
 import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.functions.Func1;
+import rx.functions.Func2;
 import rx.internal.util.RxRingBuffer;
 import rx.observers.TestSubscriber;
 import rx.schedulers.Schedulers;
@@ -655,7 +658,7 @@ public void onNext(Long t) {
     @Test
     public void testHotOperatorBackpressure() {
         TestSubscriber<String> ts = new TestSubscriber<String>();
-        Observable.timer(0, 1, TimeUnit.MILLISECONDS)
+        Observable.timer(0, 1, TimeUnit.MICROSECONDS)
                 .observeOn(Schedulers.computation())
                 .map(new Func1<Long, String>() {
 
@@ -673,6 +676,44 @@ public String call(Long t1) {
 
         ts.awaitTerminalEvent();
         System.out.println("Errors: " + ts.getOnErrorEvents());
+        assertEquals(1, ts.getOnErrorEvents().size());
+        assertEquals(MissingBackpressureException.class, ts.getOnErrorEvents().get(0).getClass());
+    }
+
+    @Test
+    public void testErrorPropagatesWhenNoOutstandingRequests() {
+        Observable<Long> timer = Observable.timer(0, 1, TimeUnit.MICROSECONDS)
+                .doOnEach(new Action1<Notification<? super Long>>() {
+
+                    @Override
+                    public void call(Notification<? super Long> n) {
+                        //                        System.out.println("BEFORE " + n);
+                    }
+
+                })
+                .observeOn(Schedulers.newThread())
+                .doOnEach(new Action1<Notification<? super Long>>() {
+
+                    @Override
+                    public void call(Notification<? super Long> n) {
+                        //                        System.out.println("AFTER " + n);
+                    }
+
+                });
+
+        TestSubscriber<Long> ts = new TestSubscriber<Long>();
+
+        Observable.combineLatest(timer, Observable.<Integer> never(), new Func2<Long, Integer, Long>() {
+
+            @Override
+            public Long call(Long t1, Integer t2) {
+                return t1;
+            }
+
+        }).take(RxRingBuffer.SIZE * 2).subscribe(ts);
+
+        ts.awaitTerminalEvent();
+        assertEquals(1, ts.getOnErrorEvents().size());
         assertEquals(MissingBackpressureException.class, ts.getOnErrorEvents().get(0).getClass());
     }