Merge pull request #1683 from benjchristensen/observeOn-error

ObserveOn Error Handling

Author: benjchristensen

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

@@ -76,7 +76,7 @@ public OnErrorSentinel(Throwable e) {
         }
 
         public String toString() {
-            return "Notification=>Error:" + e.getMessage();
+            return "Notification=>Error:" + e;
         }
     }
 

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

@@ -71,6 +71,7 @@ private static final class ObserveOnSubscriber<T> extends Subscriber<T> {
 
         private final RxRingBuffer queue = RxRingBuffer.getSpscInstance();
         private boolean completed = false;
+        private boolean failure = false;
 
         private volatile long requested = 0;
         @SuppressWarnings("rawtypes")
@@ -137,7 +138,11 @@ public void onError(final Throwable e) {
             if (isUnsubscribed() || completed) {
                 return;
             }
+            // unsubscribe eagerly since time will pass before the scheduled onError results in an unsubscribe event
+            unsubscribe();
             completed = true;
+            // mark failure so the polling thread will skip onNext still in the queue
+            failure = true;
             queue.onError(e);
             schedule();
         }
@@ -166,9 +171,18 @@ private void pollQueue() {
                             REQUESTED.incrementAndGet(this);
                             break;
                         } else {
-                            if (!on.accept(child, o)) {
-                                // non-terminal event so let's increment count
-                                emitted++;
+                            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);
+                                }
+                            } else {
+                                if (!on.accept(child, o)) {
+                                    // non-terminal event so let's increment count
+                                    emitted++;
+                                }
                             }
                         }
                     } else {

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

@@ -52,6 +52,7 @@
 import rx.observers.TestSubscriber;
 import rx.schedulers.Schedulers;
 import rx.schedulers.TestScheduler;
+import rx.subjects.PublishSubject;
 
 public class OperatorObserveOnTest {
 
@@ -602,4 +603,50 @@ public void testAsyncChild() {
         ts.awaitTerminalEvent();
         ts.assertNoErrors();
     }
+
+    @Test
+    public void testOnErrorCutsAheadOfOnNext() {
+        final PublishSubject<Long> subject = PublishSubject.create();
+
+        final AtomicLong counter = new AtomicLong();
+        TestSubscriber<Long> ts = new TestSubscriber<Long>(new Observer<Long>() {
+
+            @Override
+            public void onCompleted() {
+
+            }
+
+            @Override
+            public void onError(Throwable e) {
+
+            }
+
+            @Override
+            public void onNext(Long t) {
+                // simulate slow consumer to force backpressure failure
+                try {
+                    Thread.sleep(1);
+                } catch (InterruptedException e) {
+                }
+            }
+
+        });
+        subject.observeOn(Schedulers.computation()).subscribe(ts);
+
+        // this will blow up with backpressure
+        while (counter.get() < 102400) {
+            subject.onNext(counter.get());
+            counter.incrementAndGet();
+        }
+
+        ts.awaitTerminalEvent();
+        assertEquals(1, ts.getOnErrorEvents().size());
+        assertTrue(ts.getOnErrorEvents().get(0) instanceof MissingBackpressureException);
+        // assert that the values are sequential, that cutting in didn't allow skipping some but emitting others.
+        // example [0, 1, 2] not [0, 1, 4]
+        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);
+
+    }
 }