Change onError(Exception) to onError(Throwable)

See issue "Observer#onError should use Throwable" for background => #296

Author: benjchristensen

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

@@ -23,7 +23,7 @@
 public class Notification<T> {
 
     private final Kind kind;
-    private final Exception exception;
+    private final Throwable exception;
     private final T value;
 
     /**
@@ -44,7 +44,7 @@ public Notification(T value) {
      * @param exception
      *            The exception passed to the onError notification.
      */
-    public Notification(Exception exception) {
+    public Notification(Throwable exception) {
         this.exception = exception;
         this.value = null;
         this.kind = Kind.OnError;
@@ -62,9 +62,9 @@ public Notification() {
     /**
      * Retrieves the exception associated with an onError notification.
      * 
-     * @return The exception associated with an onError notification.
+     * @return Throwable associated with an onError notification.
      */
-    public Exception getException() {
+    public Throwable getThrowable() {
         return exception;
     }
 
@@ -126,7 +126,7 @@ public String toString() {
         if (hasValue())
             str.append(" ").append(getValue());
         if (hasException())
-            str.append(" ").append(getException().getMessage());
+            str.append(" ").append(getThrowable().getMessage());
         str.append("]");
         return str.toString();
     }
@@ -137,7 +137,7 @@ public int hashCode() {
         if (hasValue())
             hash = hash * 31 + getValue().hashCode();
         if (hasException())
-            hash = hash * 31 + getException().hashCode();
+            hash = hash * 31 + getThrowable().hashCode();
         return hash;
     }
 
@@ -154,7 +154,7 @@ public boolean equals(Object obj) {
             return false;
         if (hasValue() && !getValue().equals(notification.getValue()))
             return false;
-        if (hasException() && !getException().equals(notification.getException()))
+        if (hasException() && !getThrowable().equals(notification.getThrowable()))
             return false;
         return true;
     }

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

@@ -205,14 +205,14 @@ public Subscription subscribe(Observer<T> observer) {
         } catch (OnErrorNotImplementedException e) {
             // special handling when onError is not implemented ... we just rethrow
             throw e;
-        } catch (Exception e) {
+        } catch (Throwable e) {
             // if an unhandled error occurs executing the onSubscribe we will propagate it
             try {
                 observer.onError(hook.onSubscribeError(this, e));
             } catch (OnErrorNotImplementedException e2) {
                 // special handling when onError is not implemented ... we just rethrow
                 throw e2;
-            } catch (Exception e2) {
+            } catch (Throwable e2) {
                 // if this happens it means the onError itself failed (perhaps an invalid function implementation)
                 // so we are unable to propagate the error correctly and will just throw
                 RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2);
@@ -295,7 +295,7 @@ public void onCompleted() {
             }
 
             @Override
-            public void onError(Exception e) {
+            public void onError(Throwable e) {
                 handleError(e);
                 Object onError = callbacks.get("onError");
                 if (onError != null) {
@@ -344,7 +344,7 @@ public void onCompleted() {
             }
 
             @Override
-            public void onError(Exception e) {
+            public void onError(Throwable e) {
                 handleError(e);
                 throw new OnErrorNotImplementedException(e);
             }
@@ -379,7 +379,7 @@ public void onCompleted() {
             }
 
             @Override
-            public void onError(Exception e) {
+            public void onError(Throwable e) {
                 handleError(e);
                 throw new OnErrorNotImplementedException(e);
             }
@@ -421,7 +421,7 @@ public void onCompleted() {
             }
 
             @Override
-            public void onError(Exception e) {
+            public void onError(Throwable e) {
                 handleError(e);
                 Functions.from(onError).call(e);
             }
@@ -438,7 +438,7 @@ public Subscription subscribe(final Object onNext, final Object onError, Schedul
         return subscribeOn(scheduler).subscribe(onNext, onError);
     }
 
-    public Subscription subscribe(final Action1<T> onNext, final Action1<Exception> onError) {
+    public Subscription subscribe(final Action1<T> onNext, final Action1<Throwable> onError) {
         if (onNext == null) {
             throw new IllegalArgumentException("onNext can not be null");
         }
@@ -459,7 +459,7 @@ public void onCompleted() {
             }
 
             @Override
-            public void onError(Exception e) {
+            public void onError(Throwable e) {
                 handleError(e);
                 onError.call(e);
             }
@@ -472,7 +472,7 @@ public void onNext(T args) {
         });
     }
 
-    public Subscription subscribe(final Action1<T> onNext, final Action1<Exception> onError, Scheduler scheduler) {
+    public Subscription subscribe(final Action1<T> onNext, final Action1<Throwable> onError, Scheduler scheduler) {
         return subscribeOn(scheduler).subscribe(onNext, onError);
     }
 
@@ -504,7 +504,7 @@ public void onCompleted() {
             }
 
             @Override
-            public void onError(Exception e) {
+            public void onError(Throwable e) {
                 handleError(e);
                 Functions.from(onError).call(e);
             }
@@ -521,7 +521,7 @@ public Subscription subscribe(final Object onNext, final Object onError, final O
         return subscribeOn(scheduler).subscribe(onNext, onError, onComplete);
     }
 
-    public Subscription subscribe(final Action1<T> onNext, final Action1<Exception> onError, final Action0 onComplete) {
+    public Subscription subscribe(final Action1<T> onNext, final Action1<Throwable> onError, final Action0 onComplete) {
         if (onNext == null) {
             throw new IllegalArgumentException("onNext can not be null");
         }
@@ -545,7 +545,7 @@ public void onCompleted() {
             }
 
             @Override
-            public void onError(Exception e) {
+            public void onError(Throwable e) {
                 handleError(e);
                 onError.call(e);
             }
@@ -558,7 +558,7 @@ public void onNext(T args) {
         });
     }
 
-    public Subscription subscribe(final Action1<T> onNext, final Action1<Exception> onError, final Action0 onComplete, Scheduler scheduler) {
+    public Subscription subscribe(final Action1<T> onNext, final Action1<Throwable> onError, final Action0 onComplete, Scheduler scheduler) {
         return subscribeOn(scheduler).subscribe(onNext, onError, onComplete);
     }
 
@@ -583,7 +583,7 @@ public <R> ConnectableObservable<R> multicast(Subject<T, R> subject) {
      *
      * @param e
      */
-    private void handleError(Exception e) {
+    private void handleError(Throwable e) {
         // onError should be rare so we'll only fetch when needed
         RxJavaPlugins.getInstance().getErrorHandler().handleError(e);
     }
@@ -618,7 +618,7 @@ public Subscription call(Observer<T> t1) {
      */
     private static class ThrowObservable<T> extends Observable<T> {
 
-        public ThrowObservable(final Exception exception) {
+        public ThrowObservable(final Throwable exception) {
             super(new Func1<Observer<T>, Subscription>() {
 
                 /**
@@ -956,7 +956,7 @@ public static <T> Observable<T> empty() {
      * @return an Observable that invokes the {@link Observer}'s
      *         {@link Observer#onError onError} method when the Observer subscribes to it
      */
-    public static <T> Observable<T> error(Exception exception) {
+    public static <T> Observable<T> error(Throwable exception) {
         return new ThrowObservable<T>(exception);
     }
 
@@ -1767,7 +1767,7 @@ public static <T> Observable<T> never() {
      *            encounters an error
      * @return an Observable, identical to the source Observable with its behavior modified as described
      */
-    public static <T> Observable<T> onErrorResumeNext(final Observable<T> that, final Func1<Exception, Observable<T>> resumeFunction) {
+    public static <T> Observable<T> onErrorResumeNext(final Observable<T> that, final Func1<Throwable, Observable<T>> resumeFunction) {
         return create(OperationOnErrorResumeNextViaFunction.onErrorResumeNextViaFunction(that, resumeFunction));
     }
 
@@ -1800,11 +1800,11 @@ public static <T> Observable<T> onErrorResumeNext(final Observable<T> that, fina
     public static <T> Observable<T> onErrorResumeNext(final Observable<T> that, final Object resumeFunction) {
         @SuppressWarnings("rawtypes")
         final FuncN _f = Functions.from(resumeFunction);
-        return onErrorResumeNext(that, new Func1<Exception, Observable<T>>() {
+        return onErrorResumeNext(that, new Func1<Throwable, Observable<T>>() {
 
             @SuppressWarnings("unchecked")
             @Override
-            public Observable<T> call(Exception e) {
+            public Observable<T> call(Throwable e) {
                 return (Observable<T>) _f.call(e);
             }
         });
@@ -1866,7 +1866,7 @@ public static <T> Observable<T> onErrorResumeNext(final Observable<T> that, fina
      *            would otherwise cause it to invoke {@link Observer#onError onError}
      * @return an Observable, identical to the source Observable with its behavior modified as described
      */
-    public static <T> Observable<T> onErrorReturn(final Observable<T> that, Func1<Exception, T> resumeFunction) {
+    public static <T> Observable<T> onErrorReturn(final Observable<T> that, Func1<Throwable, T> resumeFunction) {
         return create(OperationOnErrorReturn.onErrorReturn(that, resumeFunction));
     }
 
@@ -3537,7 +3537,7 @@ public Observable<T> observeOn(Scheduler scheduler) {
      * @return an Observable that emits the items and notifications embedded in the
      *         {@link Notification} objects emitted by the source Observable
      * @see <a href="http://msdn.microsoft.com/en-us/library/hh229047(v=vs.103).aspx">MSDN: Observable.dematerialize</a>
-     * @throws Exception
+     * @throws Throwable
      *             if the source Observable is not of type {@code Observable<Notification<T>>}.
      */
     @SuppressWarnings("unchecked")
@@ -3571,7 +3571,7 @@ public <T2> Observable<T2> dematerialize() {
      *            encounters an error
      * @return the original Observable, with appropriately modified behavior
      */
-    public Observable<T> onErrorResumeNext(final Func1<Exception, Observable<T>> resumeFunction) {
+    public Observable<T> onErrorResumeNext(final Func1<Throwable, Observable<T>> resumeFunction) {
         return onErrorResumeNext(this, resumeFunction);
     }
 
@@ -3601,11 +3601,11 @@ public Observable<T> onErrorResumeNext(final Func1<Exception, Observable<T>> res
     public Observable<T> onErrorResumeNext(final Object resumeFunction) {
         @SuppressWarnings("rawtypes")
         final FuncN _f = Functions.from(resumeFunction);
-        return onErrorResumeNext(this, new Func1<Exception, Observable<T>>() {
+        return onErrorResumeNext(this, new Func1<Throwable, Observable<T>>() {
 
             @Override
             @SuppressWarnings("unchecked")
-            public Observable<T> call(Exception e) {
+            public Observable<T> call(Throwable e) {
                 return (Observable<T>) _f.call(e);
             }
         });
@@ -3664,7 +3664,7 @@ public Observable<T> onErrorResumeNext(final Observable<T> resumeSequence) {
      *            Observable encounters an error
      * @return the original Observable with appropriately modified behavior
      */
-    public Observable<T> onErrorReturn(Func1<Exception, T> resumeFunction) {
+    public Observable<T> onErrorReturn(Func1<Throwable, T> resumeFunction) {
         return onErrorReturn(this, resumeFunction);
     }
 
@@ -3694,11 +3694,11 @@ public Observable<T> onErrorReturn(Func1<Exception, T> resumeFunction) {
     public Observable<T> onErrorReturn(final Object resumeFunction) {
         @SuppressWarnings("rawtypes")
         final FuncN _f = Functions.from(resumeFunction);
-        return onErrorReturn(this, new Func1<Exception, T>() {
+        return onErrorReturn(this, new Func1<Throwable, T>() {
 
             @Override
             @SuppressWarnings("unchecked")
-            public T call(Exception e) {
+            public T call(Throwable e) {
                 return (T) _f.call(e);
             }
         });
@@ -4376,7 +4376,7 @@ public Subscription call(Observer<String> Observer) {
             verify(aObserver, times(1)).onNext("one");
             verify(aObserver, times(1)).onNext("two");
             verify(aObserver, times(1)).onNext("three");
-            verify(aObserver, Mockito.never()).onError(any(Exception.class));
+            verify(aObserver, Mockito.never()).onError(any(Throwable.class));
             verify(aObserver, times(1)).onCompleted();
         }
 
@@ -4453,7 +4453,7 @@ public void testMaterializeDematerializeChaining() {
 
             verify(observer, times(1)).onNext(1);
             verify(observer, times(1)).onCompleted();
-            verify(observer, times(0)).onError(any(Exception.class));
+            verify(observer, times(0)).onError(any(Throwable.class));
         }
 
         /**
@@ -4467,7 +4467,7 @@ public void testMaterializeDematerializeChaining() {
         public void testCustomObservableWithErrorInObserverAsynchronous() throws InterruptedException {
             final CountDownLatch latch = new CountDownLatch(1);
             final AtomicInteger count = new AtomicInteger();
-            final AtomicReference<Exception> error = new AtomicReference<Exception>();
+            final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
             Observable.create(new Func1<Observer<String>, Subscription>() {
 
                 @Override
@@ -4499,7 +4499,7 @@ public void onCompleted() {
                 }
 
                 @Override
-                public void onError(Exception e) {
+                public void onError(Throwable e) {
                     error.set(e);
                     System.out.println("error");
                     e.printStackTrace();
@@ -4533,7 +4533,7 @@ public void onNext(String v) {
         @Test
         public void testCustomObservableWithErrorInObserverSynchronous() {
             final AtomicInteger count = new AtomicInteger();
-            final AtomicReference<Exception> error = new AtomicReference<Exception>();
+            final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
             Observable.create(new Func1<Observer<String>, Subscription>() {
 
                 @Override
@@ -4553,7 +4553,7 @@ public void onCompleted() {
                 }
 
                 @Override
-                public void onError(Exception e) {
+                public void onError(Throwable e) {
                     error.set(e);
                     System.out.println("error");
                     e.printStackTrace();
@@ -4584,7 +4584,7 @@ public void onNext(String v) {
         @Test
         public void testCustomObservableWithErrorInObservableSynchronous() {
             final AtomicInteger count = new AtomicInteger();
-            final AtomicReference<Exception> error = new AtomicReference<Exception>();
+            final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
             Observable.create(new Func1<Observer<String>, Subscription>() {
 
                 @Override
@@ -4601,7 +4601,7 @@ public void onCompleted() {
                 }
 
                 @Override
-                public void onError(Exception e) {
+                public void onError(Throwable e) {
                     error.set(e);
                     System.out.println("error");
                     e.printStackTrace();
@@ -4803,7 +4803,7 @@ public void call(Object t1) {
 
                 });
                 fail("expected exception");
-            } catch (Exception e) {
+            } catch (Throwable e) {
                 assertEquals("failure", e.getMessage());
             }
         }
@@ -4822,7 +4822,7 @@ public void call(Object t1) {
         @Test
         public void testErrorThrownWithoutErrorHandlerAsynchronous() throws InterruptedException {
             final CountDownLatch latch = new CountDownLatch(1);
-            final AtomicReference<Exception> exception = new AtomicReference<Exception>();
+            final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
             Observable.create(new Func1<Observer<String>, Subscription>() {
 
                 @Override
@@ -4833,7 +4833,7 @@ public Subscription call(final Observer<String> observer) {
                         public void run() {
                             try {
                                 observer.onError(new RuntimeException("failure"));
-                            } catch (Exception e) {
+                            } catch (Throwable e) {
                                 // without an onError handler it has to just throw on whatever thread invokes it
                                 exception.set(e);
                             }

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

@@ -42,7 +42,7 @@ public interface Observer<T> {
      *
      * @param e
      */
-    public void onError(Exception e);
+    public void onError(Throwable e);
 
     /**
      * Provides the Observer with new data.