Merge pull request #3644 from akarnokd/SyncOnSubscribeErrorMgmt1x

1.x: fix SyncOnSubscribe not signalling onError if the generator crashes

Author: akarnokd

src/main/java/rx/observables/SyncOnSubscribe.java

@@ -24,6 +24,7 @@
 import rx.Subscriber;
 import rx.Subscription;
 import rx.annotations.Experimental;
+import rx.exceptions.Exceptions;
 import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.functions.Action2;
@@ -53,7 +54,16 @@ public abstract class SyncOnSubscribe<S, T> implements OnSubscribe<T> {
      */
     @Override
     public final void call(final Subscriber<? super T> subscriber) {
-        S state = generateState();
+        S state;
+        
+        try {
+            state = generateState();
+        } catch (Throwable e) {
+            Exceptions.throwIfFatal(e);
+            subscriber.onError(e);
+            return;
+        }
+        
         SubscriptionProducer<S, T> p = new SubscriptionProducer<S, T>(subscriber, this, state);
         subscriber.add(p);
         subscriber.setProducer(p);
@@ -363,7 +373,12 @@ private boolean tryUnsubscribe() {
         }
 
         private void doUnsubscribe() {
-            parent.onUnsubscribe(state);
+            try {
+                parent.onUnsubscribe(state);
+            } catch (Throwable e) {
+                Exceptions.throwIfFatal(e);
+                RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
+            }
         }
 
         @Override

src/test/java/rx/observables/SyncOnSubscribeTest.java

@@ -989,4 +989,27 @@ public Object call() throws Exception {
             if (exec != null) exec.shutdownNow();
         }
     }
+    
+    @Test
+    public void testStateThrows() {
+        TestSubscriber<Object> ts = new TestSubscriber<Object>();
+        
+        SyncOnSubscribe.<Object, Object>createSingleState(
+                new Func0<Object>() {
+                    @Override
+                    public Object call() {
+                        throw new TestException();
+                    }
+                }
+        , new Action2<Object, Observer<Object>>() {
+            @Override
+            public void call(Object s, Observer<? super Object> o) { 
+                
+            }
+        }).call(ts);
+        
+        ts.assertNoValues();
+        ts.assertError(TestException.class);
+        ts.assertNotCompleted();
+    }
 }