Merge pull request #3640 from davidmoten/fix-onBackpressureDrop-error…

…-handling fix error handling in onBackpressureDrop
ReactiveX · Jan 24, 2016 · 71d3d0f · 71d3d0f
2 parents 2dbf686 + 9675189
commit 71d3d0f
Show file tree

Hide file tree

Showing 2 changed files with 38 additions and 1 deletion.
diff --git a/src/main/java/rx/internal/operators/OperatorOnBackpressureDrop.java b/src/main/java/rx/internal/operators/OperatorOnBackpressureDrop.java
@@ -20,6 +20,7 @@
 import rx.Observable.Operator;
 import rx.Producer;
 import rx.Subscriber;
+import rx.exceptions.Exceptions;
 import rx.functions.Action1;
 
 public class OperatorOnBackpressureDrop<T> implements Operator<T, T> {
@@ -84,7 +85,12 @@ public void onNext(T t) {
                 } else {
                     // item dropped
                     if(onDrop != null) {
-                        onDrop.call(t);
+                        try {
+                            onDrop.call(t);
+                        } catch (Throwable e) {
+                            Exceptions.throwOrReport(e, child, t);
+                            return;
+                        }
                     }
                 }
             }

diff --git a/src/test/java/rx/internal/operators/OperatorOnBackpressureDropTest.java b/src/test/java/rx/internal/operators/OperatorOnBackpressureDropTest.java
@@ -16,8 +16,10 @@
 package rx.internal.operators;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
@@ -26,6 +28,8 @@
 import rx.Observable.OnSubscribe;
 import rx.Observer;
 import rx.Subscriber;
+import rx.functions.Action0;
+import rx.functions.Action1;
 import rx.internal.util.RxRingBuffer;
 import rx.observers.TestSubscriber;
 import rx.schedulers.Schedulers;
@@ -117,6 +121,33 @@ public void onNext(Long t) {
             }});
         assertEquals(n, count.get());
     }
+
+    @Test
+    public void testNonFatalExceptionFromOverflowActionIsNotReportedFromUpstreamOperator() {
+        final AtomicBoolean errorOccurred = new AtomicBoolean(false);
+        //request 0 
+        TestSubscriber<Long> ts = TestSubscriber.create(0);
+        //range method emits regardless of requests so should trigger onBackpressureDrop action
+        range(2)
+          // if haven't caught exception in onBackpressureDrop operator then would incorrectly
+          // be picked up by this call to doOnError
+          .doOnError(new Action1<Throwable>() {
+                @Override
+                public void call(Throwable t) {
+                    errorOccurred.set(true);
+                }
+            })
+          .onBackpressureDrop(THROW_NON_FATAL)
+          .subscribe(ts);
+        assertFalse(errorOccurred.get());
+    }
+
+    private static final Action1<Long> THROW_NON_FATAL = new Action1<Long>() {
+        @Override
+        public void call(Long n) {
+            throw new RuntimeException();
+        }
+    }; 
 
     static final Observable<Long> infinite = Observable.create(new OnSubscribe<Long>() {