Unhandled errors go to UncaughtExceptionHandler

Rather than swallowing/logging errors, ScheduledAction now delivers them
to the UncaughtExceptionHandler for the executing Thread. This gives
client applications control over the handling of errors that occur off
of the calling Thread.

Author: loganj

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++;
+        }
+    }
+}