Refactored exception reporting of most operators.

Author: akarnokd

src/main/java/rx/exceptions/Exceptions.java

@@ -15,10 +15,9 @@
  */
 package rx.exceptions;
 
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 
+import rx.Observer;
 import rx.annotations.Experimental;
 
 /**
@@ -178,4 +177,28 @@ public static void throwIfAny(List<? extends Throwable> exceptions) {
                     "Multiple exceptions", exceptions);
         }
     }
+    
+    /**
+     * Forwards a fatal exception or reports it along with the value
+     * caused it to the given Observer.
+     * @param t the exception
+     * @param o the observer to report to
+     * @param value the value that caused the exception
+     */
+    @Experimental
+    public static void throwOrReport(Throwable t, Observer<?> o, Object value) {
+        Exceptions.throwIfFatal(t);
+        o.onError(OnErrorThrowable.addValueAsLastCause(t, value));
+    }
+    /**
+     * Forwards a fatal exception or reports it to the given Observer.
+     * @param t the exception
+     * @param o the observer to report to
+     * @param value the value that caused the exception
+     */
+    @Experimental
+    public static void throwOrReport(Throwable t, Observer<?> o) {
+        Exceptions.throwIfFatal(t);
+        o.onError(t);
+    }
 }

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

@@ -23,9 +23,9 @@
 
 import rx.Observable;
 import rx.Observable.OnSubscribe;
+import rx.exceptions.*;
 import rx.Producer;
 import rx.Subscriber;
-import rx.exceptions.MissingBackpressureException;
 import rx.functions.FuncN;
 import rx.internal.util.RxRingBuffer;
 
@@ -202,7 +202,7 @@ public boolean onNext(int index, T t) {
                     } catch (MissingBackpressureException e) {
                         onError(e);
                     } catch (Throwable e) {
-                        onError(e);
+                        Exceptions.throwOrReport(e, child);
                     }
                 }
             }

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

@@ -18,6 +18,7 @@
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.Subscriber;
+import rx.exceptions.Exceptions;
 import rx.functions.Func0;
 import rx.observers.Subscribers;
 
@@ -44,7 +45,7 @@ public void call(final Subscriber<? super T> s) {
         try {
             o = observableFactory.call();
         } catch (Throwable t) {
-            s.onError(t);
+            Exceptions.throwOrReport(t, s);
             return;
         }
         o.unsafeSubscribe(Subscribers.wrap(s));

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

@@ -17,6 +17,7 @@
 
 import rx.*;
 import rx.Observable.OnSubscribe;
+import rx.exceptions.Exceptions;
 import rx.functions.Func0;
 import rx.observers.Subscribers;
 
@@ -58,7 +59,7 @@ public void onNext(U t) {
 
             });
         } catch (Throwable e) {
-            child.onError(e);
+            Exceptions.throwOrReport(e, child);
         }
     }
 

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

@@ -15,24 +15,17 @@
  */
 package rx.internal.operators;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
+import rx.*;
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.Observer;
-import rx.Subscriber;
-import rx.Subscription;
-import rx.functions.Func1;
-import rx.functions.Func2;
-import rx.observers.SerializedObserver;
-import rx.observers.SerializedSubscriber;
-import rx.subjects.PublishSubject;
-import rx.subjects.Subject;
-import rx.subscriptions.CompositeSubscription;
-import rx.subscriptions.RefCountSubscription;
+import rx.exceptions.Exceptions;
+import rx.functions.*;
+import rx.observers.*;
+import rx.subjects.*;
+import rx.subscriptions.*;
 
 /**
  * Corrrelates two sequences when they overlap and groups the results.
@@ -192,7 +185,7 @@ public void onNext(T1 args) {
 
 
                 } catch (Throwable t) {
-                    onError(t);
+                    Exceptions.throwOrReport(t, this);
                 }
             }
 
@@ -242,7 +235,7 @@ public void onNext(T2 args) {
                         o.onNext(args);
                     }
                 } catch (Throwable t) {
-                    onError(t);
+                    Exceptions.throwOrReport(t, this);
                 }
             }
 

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

@@ -15,20 +15,15 @@
  */
 package rx.internal.operators;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
+import rx.*;
 import rx.Observable;
 import rx.Observable.OnSubscribe;
-import rx.Subscriber;
-import rx.Subscription;
-import rx.functions.Func1;
-import rx.functions.Func2;
+import rx.exceptions.Exceptions;
+import rx.functions.*;
 import rx.observers.SerializedSubscriber;
-import rx.subscriptions.CompositeSubscription;
-import rx.subscriptions.SerialSubscription;
+import rx.subscriptions.*;
 
 /**
  * Correlates the elements of two sequences based on overlapping durations.
@@ -154,7 +149,7 @@ public void onNext(TLeft args) {
                         subscriber.onNext(result);
                     }
                 } catch (Throwable t) {
-                    onError(t);
+                    Exceptions.throwOrReport(t, this);
                 }
             }
 
@@ -266,7 +261,7 @@ public void onNext(TRight args) {
                     }
 
                 } catch (Throwable t) {
-                    onError(t);
+                    Exceptions.throwOrReport(t, this);
                 }
             }
 

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

@@ -19,6 +19,7 @@
 import rx.Observable.OnSubscribe;
 import rx.Scheduler;
 import rx.Scheduler.Worker;
+import rx.exceptions.Exceptions;
 import rx.Subscriber;
 import rx.functions.Action0;
 
@@ -47,7 +48,7 @@ public void call() {
                 try {
                     child.onNext(0L);
                 } catch (Throwable t) {
-                    child.onError(t);
+                    Exceptions.throwOrReport(t, child);
                     return;
                 }
                 child.onCompleted();

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

@@ -19,6 +19,7 @@
 import rx.Observable.OnSubscribe;
 import rx.Scheduler;
 import rx.Scheduler.Worker;
+import rx.exceptions.Exceptions;
 import rx.Subscriber;
 import rx.functions.Action0;
 
@@ -51,9 +52,9 @@ public void call() {
                     child.onNext(counter++);
                 } catch (Throwable e) {
                     try {
-                        child.onError(e);
-                    } finally {
                         worker.unsubscribe();
+                    } finally {
+                        Exceptions.throwOrReport(e, child);
                     }
                 }
             }

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

@@ -19,6 +19,7 @@
 import java.util.concurrent.TimeUnit;
 
 import rx.Observable.OnSubscribe;
+import rx.exceptions.Exceptions;
 import rx.Subscriber;
 import rx.functions.Action0;
 import rx.subscriptions.Subscriptions;
@@ -83,7 +84,7 @@ public void call() {
                     //refuse to emit onError if already unsubscribed
                     return;
                 }
-                subscriber.onError(e);
+                Exceptions.throwOrReport(e, subscriber);
             }
         }
     }

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

@@ -20,7 +20,7 @@
 
 import rx.*;
 import rx.Observable.OnSubscribe;
-import rx.exceptions.CompositeException;
+import rx.exceptions.*;
 import rx.functions.*;
 import rx.observers.Subscribers;
 
@@ -72,6 +72,8 @@ public void call(final Subscriber<? super T> subscriber) {
                 observable.unsafeSubscribe(Subscribers.wrap(subscriber));
             } catch (Throwable e) {
                 Throwable disposeError = disposeEagerlyIfRequested(disposeOnceOnly);
+                Exceptions.throwIfFatal(e);
+                Exceptions.throwIfFatal(disposeError);
                 if (disposeError != null)
                     subscriber.onError(new CompositeException(Arrays.asList(e, disposeError)));
                 else
@@ -80,7 +82,7 @@ public void call(final Subscriber<? super T> subscriber) {
             }
         } catch (Throwable e) {
             // then propagate error
-            subscriber.onError(e);
+            Exceptions.throwOrReport(e, subscriber);
         }
     }