Merge pull request #3733 from loganj/completable-errors

Avoid swallowing errors in Completable
ReactiveX · Mar 2, 2016 · 423172f · 423172f
2 parents 02e6903 + 8b55303
commit 423172f
Show file tree

Hide file tree

Showing 4 changed files with 109 additions and 17 deletions.
diff --git a/src/main/java/rx/Completable.java b/src/main/java/rx/Completable.java
@@ -1835,6 +1835,7 @@ public void onCompleted() {
             public void onError(Throwable e) {
                 ERROR_HANDLER.handleError(e);
                 mad.unsubscribe();
+                deliverUncaughtException(e);
             }
 
             @Override
@@ -1864,14 +1865,17 @@ public void onCompleted() {
                     onComplete.call();
                 } catch (Throwable e) {
                     ERROR_HANDLER.handleError(e);
+                    deliverUncaughtException(e);
+                } finally {
+                    mad.unsubscribe();
                 }
-                mad.unsubscribe();
             }
 
             @Override
             public void onError(Throwable e) {
                 ERROR_HANDLER.handleError(e);
                 mad.unsubscribe();
+                deliverUncaughtException(e);
             }
 
             @Override
@@ -1915,8 +1919,10 @@ public void onError(Throwable e) {
                 } catch (Throwable ex) {
                     e = new CompositeException(Arrays.asList(e, ex));
                     ERROR_HANDLER.handleError(e);
+                    deliverUncaughtException(e);
+                } finally {
+                    mad.unsubscribe();
                 }
-                mad.unsubscribe();
             }
 
             @Override
@@ -1927,7 +1933,12 @@ public void onSubscribe(Subscription d) {
 
         return mad;
     }
-
+
+    private static void deliverUncaughtException(Throwable e) {
+        Thread thread = Thread.currentThread();
+        thread.getUncaughtExceptionHandler().uncaughtException(thread, e);
+    }
+
     /**
      * Subscribes the given CompletableSubscriber to this Completable instance.
      * @param s the CompletableSubscriber, not null

diff --git a/src/test/java/rx/CapturingUncaughtExceptionHandler.java b/src/test/java/rx/CapturingUncaughtExceptionHandler.java
@@ -0,0 +1,16 @@
+package rx;
+
+import java.util.concurrent.CountDownLatch;
+
+public final class CapturingUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
+    public int count = 0;
+    public Throwable caught;
+    public CountDownLatch completed = new CountDownLatch(1);
+
+    @Override
+    public void uncaughtException(Thread t, Throwable e) {
+        count++;
+        caught = e;
+        completed.countDown();
+    }
+}
diff --git a/src/test/java/rx/CompletableTest.java b/src/test/java/rx/CompletableTest.java
@@ -2700,7 +2700,64 @@ public void call(CompletableSubscriber s) {
 
         Assert.assertTrue(name.get().startsWith("RxComputation"));
     }
-
+
+    @Test
+    public void subscribeEmptyOnError() {
+        expectUncaughtTestException(new Action0() {
+            @Override public void call() {
+                error.completable.subscribe();
+            }
+        });
+    }
+
+    @Test
+    public void subscribeOneActionOnError() {
+        expectUncaughtTestException(new Action0() {
+            @Override
+            public void call() {
+                error.completable.subscribe(new Action0() {
+                    @Override
+                    public void call() {
+                    }
+                });
+            }
+        });
+    }
+
+    @Test
+    public void subscribeOneActionThrowFromOnCompleted() {
+        expectUncaughtTestException(new Action0() {
+            @Override
+            public void call() {
+                normal.completable.subscribe(new Action0() {
+                    @Override
+                    public void call() {
+                        throw new TestException();
+                    }
+                });
+            }
+        });
+    }
+
+    @Test
+    public void subscribeTwoActionsThrowFromOnError() {
+        expectUncaughtTestException(new Action0() {
+            @Override
+            public void call() {
+                error.completable.subscribe(new Action1<Throwable>() {
+                    @Override
+                    public void call(Throwable throwable) {
+                        throw new TestException();
+                    }
+                }, new Action0() {
+                    @Override
+                    public void call() {
+                    }
+                });
+            }
+        });
+    }
+
     @Test(timeout = 1000)
     public void timeoutEmitError() {
         Throwable e = Completable.never().timeout(100, TimeUnit.MILLISECONDS).get();
@@ -3742,4 +3799,24 @@ public void call(Throwable e) {
         assertNotNull("Unsubscribed before the call to onError", subscriptionRef.get());
     }
 
+    private static void expectUncaughtTestException(Action0 action) {
+        Thread.UncaughtExceptionHandler originalHandler = Thread.getDefaultUncaughtExceptionHandler();
+        CapturingUncaughtExceptionHandler handler = new CapturingUncaughtExceptionHandler();
+        Thread.setDefaultUncaughtExceptionHandler(handler);
+        try {
+            action.call();
+            assertEquals("Should have received exactly 1 exception", 1, handler.count);
+            Throwable caught = handler.caught;
+            while (caught != null) {
+                if (caught instanceof TestException) break;
+                if (caught == caught.getCause()) break;
+                caught = caught.getCause();
+            }
+            assertTrue("A TestException should have been delivered to the handler",
+                    caught instanceof TestException);
+        } finally {
+            Thread.setDefaultUncaughtExceptionHandler(originalHandler);
+        }
+    }
+
 }
diff --git a/src/test/java/rx/schedulers/SchedulerTests.java b/src/test/java/rx/schedulers/SchedulerTests.java
@@ -1,5 +1,6 @@
 package rx.schedulers;
 
+import rx.CapturingUncaughtExceptionHandler;
 import rx.Observable;
 import rx.Observer;
 import rx.Scheduler;
@@ -87,19 +88,6 @@ static void testHandledErrorIsNotDeliveredToThreadHandler(Scheduler scheduler) t
         }
     }
 
-    private static final class CapturingUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
-        int count = 0;
-        Throwable caught;
-        CountDownLatch completed = new CountDownLatch(1);
-
-        @Override
-        public void uncaughtException(Thread t, Throwable e) {
-            count++;
-            caught = e;
-            completed.countDown();
-        }
-    }
-
     private static final class CapturingObserver<T> implements Observer<T> {
         CountDownLatch completed = new CountDownLatch(1);
         int errorCount = 0;