Merge pull request #3626 from davidmoten/simplify-throw

use Exceptions.throwOrError to simplify error handling

Author: zsxwing

src/main/java/rx/Single.java

@@ -185,17 +185,15 @@ public void call(Subscriber<? super R> o) {
                         st.onStart();
                         onSubscribe.call(st);
                     } catch (Throwable e) {
-                        Exceptions.throwIfFatal(e);
                         // localized capture of errors rather than it skipping all operators
                         // and ending up in the try/catch of the subscribe method which then
                         // prevents onErrorResumeNext and other similar approaches to error handling
-                        st.onError(e);
+                        Exceptions.throwOrReport(e, st);
                     }
                 } catch (Throwable e) {
-                    Exceptions.throwIfFatal(e);
                     // if the lift function failed all we can do is pass the error to the final Subscriber
                     // as we don't have the operator available to us
-                    o.onError(e);
+                    Exceptions.throwOrReport(e, o);
                 }
             }
         });

src/main/java/rx/internal/operators/OnSubscribeFromCallable.java

@@ -31,8 +31,7 @@ public void call(Subscriber<? super T> subscriber) {
         try {
             singleDelayedProducer.setValue(resultFactory.call());
         } catch (Throwable t) {
-            Exceptions.throwIfFatal(t);
-            subscriber.onError(t);
+            Exceptions.throwOrReport(t, subscriber);
         }
     }
 }

src/main/java/rx/internal/operators/OperatorScan.java

@@ -108,8 +108,7 @@ public void onNext(T t) {
                         try {
                             v = accumulator.call(v, t);
                         } catch (Throwable e) {
-                            Exceptions.throwIfFatal(e);
-                            child.onError(OnErrorThrowable.addValueAsLastCause(e, t));
+                            Exceptions.throwOrReport(e, child, t);
                             return;
                         }
                     }
@@ -138,8 +137,7 @@ public void onNext(T currentValue) {
                 try {
                     v = accumulator.call(v, currentValue);
                 } catch (Throwable e) {
-                    Exceptions.throwIfFatal(e);
-                    onError(OnErrorThrowable.addValueAsLastCause(e, currentValue));
+                    Exceptions.throwOrReport(e, this, currentValue);
                     return;
                 }
                 value = v;
@@ -322,8 +320,7 @@ void emitLoop() {
                     try {
                         child.onNext(v);
                     } catch (Throwable ex) {
-                        Exceptions.throwIfFatal(ex);
-                        child.onError(OnErrorThrowable.addValueAsLastCause(ex, v));
+                        Exceptions.throwOrReport(ex, child, v);
                         return;
                     }
                     r--;

src/main/java/rx/internal/operators/OperatorToMap.java

@@ -83,8 +83,7 @@ public Subscriber<? super T> call(final Subscriber<? super Map<K, V>> subscriber
         try {
             localMap = mapFactory.call();
         } catch (Throwable ex) {
-            Exceptions.throwIfFatal(ex);
-            subscriber.onError(ex);
+            Exceptions.throwOrReport(ex, subscriber);
             Subscriber<? super T> parent = Subscribers.empty();
             parent.unsubscribe();
             return parent;
@@ -110,8 +109,7 @@ public void onNext(T v) {
                     key = keySelector.call(v);
                     value = valueSelector.call(v);
                 } catch (Throwable ex) {
-                    Exceptions.throwIfFatal(ex);
-                    subscriber.onError(ex);
+                    Exceptions.throwOrReport(ex, subscriber);
                     return;
                 }
 

src/main/java/rx/internal/operators/OperatorToMultimap.java

@@ -138,8 +138,7 @@ public void onNext(T v) {
                     key = keySelector.call(v);
                     value = valueSelector.call(v);
                 } catch (Throwable ex) {
-                    Exceptions.throwIfFatal(ex);
-                    subscriber.onError(ex);
+                    Exceptions.throwOrReport(ex, subscriber);
                     return;
                 }
 
@@ -148,8 +147,7 @@ public void onNext(T v) {
                     try {
                         collection = collectionFactory.call(key);
                     } catch (Throwable ex) {
-                        Exceptions.throwIfFatal(ex);
-                        subscriber.onError(ex);
+                        Exceptions.throwOrReport(ex, subscriber);
                         return;
                     }
                     map.put(key, collection);

src/main/java/rx/internal/operators/UnicastSubject.java

@@ -154,8 +154,7 @@ public void onNext(T t) {
                 try {
                     s.onNext(t);
                 } catch (Throwable ex) {
-                    Exceptions.throwIfFatal(ex);
-                    s.onError(OnErrorThrowable.addValueAsLastCause(ex, t));
+                    Exceptions.throwOrReport(ex, s, t);
                 }
             }
         }

src/main/java/rx/observers/SafeSubscriber.java

@@ -141,9 +141,7 @@ public void onNext(T args) {
         } catch (Throwable e) {
             // we handle here instead of another method so we don't add stacks to the frame
             // which can prevent it from being able to handle StackOverflow
-            Exceptions.throwIfFatal(e);
-            // handle errors if the onNext implementation fails, not just if the Observable fails
-            onError(e);
+            Exceptions.throwOrReport(e, this);
         }
     }
 

src/main/java/rx/observers/SerializedObserver.java

@@ -95,8 +95,7 @@ public void onNext(T t) {
             actual.onNext(t);
         } catch (Throwable e) {
             terminated = true;
-            Exceptions.throwIfFatal(e);
-            actual.onError(OnErrorThrowable.addValueAsLastCause(e, t));
+            Exceptions.throwOrReport(e, actual, t);
             return;
         }
         for (;;) {