Merge pull request #1631 from benjchristensen/doOnEach-error-swallowing

Handle Fatal Exceptions

Author: benjchristensen

rxjava-core/src/main/java/rx/internal/operators/OperatorDoOnEach.java

@@ -18,6 +18,7 @@
 import rx.Observable.Operator;
 import rx.Observer;
 import rx.Subscriber;
+import rx.exceptions.Exceptions;
 import rx.exceptions.OnErrorThrowable;
 
 /**
@@ -54,6 +55,8 @@ public void onCompleted() {
 
             @Override
             public void onError(Throwable e) {
+                // need to throwIfFatal since we swallow errors after terminated
+                Exceptions.throwIfFatal(e);
                 if (done) {
                     return;
                 }

rxjava-core/src/test/java/rx/internal/operators/OperatorDoOnEachTest.java

@@ -15,7 +15,7 @@
  */
 package rx.internal.operators;
 
-import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.*;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -28,6 +28,8 @@
 
 import rx.Observable;
 import rx.Observer;
+import rx.Subscriber;
+import rx.exceptions.OnErrorNotImplementedException;
 import rx.functions.Action1;
 import rx.functions.Func1;
 
@@ -121,7 +123,7 @@ public void call(String s) {
     @Test
     public void testIssue1451Case1() {
         // https://github.com/Netflix/RxJava/issues/1451
-        int[] nums = {1, 2, 3};
+        int[] nums = { 1, 2, 3 };
         final AtomicInteger count = new AtomicInteger();
         for (final int n : nums) {
             Observable
@@ -147,7 +149,7 @@ public void call(List<Boolean> booleans) {
     @Test
     public void testIssue1451Case2() {
         // https://github.com/Netflix/RxJava/issues/1451
-        int[] nums = {1, 2, 3};
+        int[] nums = { 1, 2, 3 };
         final AtomicInteger count = new AtomicInteger();
         for (final int n : nums) {
             Observable
@@ -169,4 +171,34 @@ public void call(List<Boolean> booleans) {
         }
         assertEquals(nums.length, count.get());
     }
+
+    @Test
+    public void testFatalError() {
+        try {
+            Observable.just(1, 2, 3)
+                    .flatMap(new Func1<Integer, Observable<?>>() {
+                        @Override
+                        public Observable<?> call(Integer integer) {
+                            return Observable.create(new Observable.OnSubscribe<Object>() {
+                                @Override
+                                public void call(Subscriber<Object> o) {
+                                    throw new NullPointerException("Test NPE");
+                                }
+                            });
+                        }
+                    })
+                    .doOnNext(new Action1<Object>() {
+                        @Override
+                        public void call(Object o) {
+                            System.out.println("Won't come here");
+                        }
+                    })
+                    .subscribe();
+            fail("should have thrown an exception");
+        } catch (OnErrorNotImplementedException e) {
+            assertTrue(e.getCause() instanceof NullPointerException);
+            assertEquals(e.getCause().getMessage(), "Test NPE");
+            System.out.println("Received exception: " + e);
+        }
+    }
 }