Merge pull request #1091 from benjchristensen/unsafeSubscribe-error-handling

Handle Thrown Errors with UnsafeSubscribe

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -6231,7 +6231,15 @@ public void onNext(T t) {
      * @return Subscription which is the Subscriber passed in
      */
     public final Subscription unsafeSubscribe(Subscriber<? super T> subscriber) {
-        onSubscribe.call(subscriber);
+        try {
+            onSubscribe.call(subscriber);
+        } catch (Throwable e) {
+            if (e instanceof OnErrorNotImplementedException) {
+                throw (OnErrorNotImplementedException) e;
+            }
+            // handle broken contracts: https://github.com/Netflix/RxJava/issues/1090
+            subscriber.onError(e);
+        }
         return subscriber;
     }
 

rxjava-core/src/test/java/rx/operators/OperatorMergeTest.java

@@ -323,6 +323,24 @@ public void testError2() {
         verify(stringObserver, times(0)).onNext("eight");
         verify(stringObserver, times(0)).onNext("nine");
     }
+    
+    @Test
+    public void testThrownErrorHandling() {
+        TestSubscriber<String> ts = new TestSubscriber<String>();
+        Observable<String> o1 = Observable.create(new OnSubscribe<String>() {
+
+            @Override
+            public void call(Subscriber<? super String> s) {
+                throw new RuntimeException("fail");
+            }
+
+        });
+
+        Observable.merge(o1, o1).subscribe(ts);
+        ts.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS);
+        ts.assertTerminalEvent();
+        System.out.println("Error: " + ts.getOnErrorEvents());
+    }
 
     private static class TestSynchronousObservable implements Observable.OnSubscribeFunc<String> {
 

rxjava-core/src/test/java/rx/operators/OperatorSubscribeOnTest.java

@@ -77,6 +77,36 @@ public void call(
         assertEquals(0, observer.getOnErrorEvents().size());
         assertEquals(1, observer.getOnCompletedEvents().size());
     }
+    
+    @Test
+    public void testThrownErrorHandling() {
+        TestSubscriber<String> ts = new TestSubscriber<String>();
+        Observable.create(new OnSubscribe<String>() {
+
+            @Override
+            public void call(Subscriber<? super String> s) {
+                throw new RuntimeException("fail");
+            }
+
+        }).subscribeOn(Schedulers.computation()).subscribe(ts);
+        ts.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS);
+        ts.assertTerminalEvent();
+    }
+    
+    @Test
+    public void testOnError() {
+        TestSubscriber<String> ts = new TestSubscriber<String>();
+        Observable.create(new OnSubscribe<String>() {
+
+            @Override
+            public void call(Subscriber<? super String> s) {
+                s.onError(new RuntimeException("fail"));
+            }
+
+        }).subscribeOn(Schedulers.computation()).subscribe(ts);
+        ts.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS);
+        ts.assertTerminalEvent();
+    }
 
     public static class SlowScheduler extends Scheduler {
         final Scheduler actual;