Merge pull request #1766 from loganj/uncaught

Unhandled errors go to UncaughtExceptionHandler

Author: benjchristensen

src/main/java/rx/internal/schedulers/ScheduledAction.java

@@ -51,8 +51,9 @@ public void run() {
             } else {
                 ie = new IllegalStateException("Fatal Exception thrown on Scheduler.Worker thread.", e);
             }
-            ie.printStackTrace();
             RxJavaPlugins.getInstance().getErrorHandler().handleError(ie);
+            Thread thread = Thread.currentThread();
+            thread.getUncaughtExceptionHandler().uncaughtException(thread, ie);
         } finally {
             unsubscribe();
         }

src/main/java/rx/schedulers/Schedulers.java

@@ -75,7 +75,9 @@ public static Scheduler trampoline() {
 
     /**
      * Creates and returns a {@link Scheduler} that creates a new {@link Thread} for each unit of work.
-     * 
+     * <p>
+     * Unhandled errors will be delivered to the scheduler Thread's {@link java.lang.Thread.UncaughtExceptionHandler}.
+     *
      * @return a {@link NewThreadScheduler} instance
      */
     public static Scheduler newThread() {
@@ -88,7 +90,9 @@ public static Scheduler newThread() {
      * This can be used for event-loops, processing callbacks and other computational work.
      * <p>
      * Do not perform IO-bound work on this scheduler. Use {@link #io()} instead.
-     * 
+     * <p>
+     * Unhandled errors will be delivered to the scheduler Thread's {@link java.lang.Thread.UncaughtExceptionHandler}.
+     *
      * @return a {@link Scheduler} meant for computation-bound work
      */
     public static Scheduler computation() {
@@ -103,7 +107,9 @@ public static Scheduler computation() {
      * This can be used for asynchronously performing blocking IO.
      * <p>
      * Do not perform computational work on this scheduler. Use {@link #computation()} instead.
-     * 
+     * <p>
+     * Unhandled errors will be delivered to the scheduler Thread's {@link java.lang.Thread.UncaughtExceptionHandler}.
+     *
      * @return a {@link Scheduler} meant for IO-bound work
      */
     public static Scheduler io() {

src/test/java/rx/schedulers/CachedThreadSchedulerTest.java

@@ -57,4 +57,13 @@ public void call(String t) {
         });
     }
 
+    @Test
+    public final void testUnhandledErrorIsDeliveredToThreadHandler() throws InterruptedException {
+        SchedulerTests.testUnhandledErrorIsDeliveredToThreadHandler(getScheduler());
+    }
+
+    @Test
+    public final void testHandledErrorIsNotDeliveredToThreadHandler() throws InterruptedException {
+        SchedulerTests.testHandledErrorIsNotDeliveredToThreadHandler(getScheduler());
+    }
 }

src/test/java/rx/schedulers/ComputationSchedulerTests.java

@@ -137,4 +137,14 @@ public void call(String t) {
             }
         });
     }
+
+    @Test
+    public final void testUnhandledErrorIsDeliveredToThreadHandler() throws InterruptedException {
+        SchedulerTests.testUnhandledErrorIsDeliveredToThreadHandler(getScheduler());
+    }
+
+    @Test
+    public final void testHandledErrorIsNotDeliveredToThreadHandler() throws InterruptedException {
+        SchedulerTests.testHandledErrorIsNotDeliveredToThreadHandler(getScheduler());
+    }
 }

src/test/java/rx/schedulers/NewThreadSchedulerTest.java

@@ -16,6 +16,7 @@
 
 package rx.schedulers;
 
+import org.junit.Test;
 import rx.Scheduler;
 
 public class NewThreadSchedulerTest extends AbstractSchedulerConcurrencyTests {
@@ -24,4 +25,14 @@ public class NewThreadSchedulerTest extends AbstractSchedulerConcurrencyTests {
     protected Scheduler getScheduler() {
         return Schedulers.newThread();
     }
+
+    @Test
+    public final void testUnhandledErrorIsDeliveredToThreadHandler() throws InterruptedException {
+        SchedulerTests.testUnhandledErrorIsDeliveredToThreadHandler(getScheduler());
+    }
+
+    @Test
+    public final void testHandledErrorIsNotDeliveredToThreadHandler() throws InterruptedException {
+        SchedulerTests.testHandledErrorIsNotDeliveredToThreadHandler(getScheduler());
+    }
 }

src/test/java/rx/schedulers/SchedulerTests.java

@@ -0,0 +1,125 @@
+package rx.schedulers;
+
+import rx.Observable;
+import rx.Observer;
+import rx.Scheduler;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+final class SchedulerTests {
+    private SchedulerTests() {
+        // No instances.
+    }
+
+    /**
+     * Verifies that the given Scheduler delivers unhandled errors to its executing thread's
+     * {@link java.lang.Thread.UncaughtExceptionHandler}.
+     * <p>
+     * Schedulers which execute on a separate thread from their calling thread should exhibit this behavior. Schedulers
+     * which execute on their calling thread may not.
+     */
+    static void testUnhandledErrorIsDeliveredToThreadHandler(Scheduler scheduler) throws InterruptedException {
+        Thread.UncaughtExceptionHandler originalHandler = Thread.getDefaultUncaughtExceptionHandler();
+        try {
+            CapturingUncaughtExceptionHandler handler = new CapturingUncaughtExceptionHandler();
+            Thread.setDefaultUncaughtExceptionHandler(handler);
+            IllegalStateException error = new IllegalStateException("Should be delivered to handler");
+            Observable.error(error)
+                    .subscribeOn(scheduler)
+                    .subscribe();
+
+            if (!handler.completed.await(3, TimeUnit.SECONDS)) {
+                fail("timed out");
+            }
+
+            assertEquals("Should have received exactly 1 exception", 1, handler.count);
+            Throwable cause = handler.caught;
+            while (cause != null) {
+                if (error.equals(cause)) break;
+                if (cause == cause.getCause()) break;
+                cause = cause.getCause();
+            }
+            assertEquals("Our error should have been delivered to the handler", error, cause);
+        } finally {
+            Thread.setDefaultUncaughtExceptionHandler(originalHandler);
+        }
+    }
+
+    /**
+     * Verifies that the given Scheduler does not deliver handled errors to its executing Thread's
+     * {@link java.lang.Thread.UncaughtExceptionHandler}.
+     * <p>
+     * This is a companion test to {@link #testUnhandledErrorIsDeliveredToThreadHandler}, and is needed only for the
+     * same Schedulers.
+     */
+    static void testHandledErrorIsNotDeliveredToThreadHandler(Scheduler scheduler) throws InterruptedException {
+        Thread.UncaughtExceptionHandler originalHandler = Thread.getDefaultUncaughtExceptionHandler();
+        try {
+            CapturingUncaughtExceptionHandler handler = new CapturingUncaughtExceptionHandler();
+            CapturingObserver<Object> observer = new CapturingObserver<Object>();
+            Thread.setDefaultUncaughtExceptionHandler(handler);
+            IllegalStateException error = new IllegalStateException("Should be delivered to handler");
+            Observable.error(error)
+                    .subscribeOn(scheduler)
+                    .subscribe(observer);
+
+            if (!observer.completed.await(3, TimeUnit.SECONDS)) {
+                fail("timed out");
+            }
+
+            assertEquals("Handler should not have received anything", 0, handler.count);
+            assertEquals("Observer should have received an error", 1, observer.errorCount);
+            assertEquals("Observer should not have received a next value", 0, observer.nextCount);
+
+            Throwable cause = observer.error;
+            while (cause != null) {
+                if (error.equals(cause)) break;
+                if (cause == cause.getCause()) break;
+                cause = cause.getCause();
+            }
+            assertEquals("Our error should have been delivered to the observer", error, cause);
+        } finally {
+            Thread.setDefaultUncaughtExceptionHandler(originalHandler);
+        }
+    }
+
+    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;
+        int nextCount = 0;
+        Throwable error;
+
+        @Override
+        public void onCompleted() {
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            errorCount++;
+            error = e;
+            completed.countDown();
+        }
+
+        @Override
+        public void onNext(T t) {
+            nextCount++;
+        }
+    }
+}