Merge pull request ReactiveX#3408 from akarnokd/DoOnErrorReportBoth

DoOnEach: report both original exception and callback exception.

Author: abersnaze

src/main/java/rx/Observable.java

@@ -4362,6 +4362,10 @@ public final void onNext(T v) {
     /**
      * Modifies the source Observable so that it notifies an Observer for each item it emits.
      * <p>
+     * In case the onError of the supplied observer throws, the downstream will receive a composite exception containing
+     * the original exception and the exception thrown by onError. If the onNext or the onCompleted methods
+     * of the supplied observer throws, the downstream will be terminated and wil receive this thrown exception.
+     * <p>
      * <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/doOnEach.png" alt="">
      * <dl>
      *  <dt><b>Scheduler:</b></dt>
@@ -4380,6 +4384,9 @@ public final Observable<T> doOnEach(Observer<? super T> observer) {
     /**
      * Modifies the source Observable so that it invokes an action if it calls {@code onError}.
      * <p>
+     * In case the onError action throws, the downstream will receive a composite exception containing
+     * the original exception and the exception thrown by onError.
+     * <p>
      * <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/doOnError.png" alt="">
      * <dl>
      *  <dt><b>Scheduler:</b></dt>

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

@@ -15,9 +15,11 @@
  */
 package rx.internal.operators;
 
+import java.util.Arrays;
+
 import rx.*;
 import rx.Observable.Operator;
-import rx.exceptions.Exceptions;
+import rx.exceptions.*;
 
 /**
  * Converts the elements of an observable sequence to the specified type.
@@ -62,7 +64,8 @@ public void onError(Throwable e) {
                 try {
                     doOnEachObserver.onError(e);
                 } catch (Throwable e2) {
-                    Exceptions.throwOrReport(e2, observer);
+                    Exceptions.throwIfFatal(e2);
+                    observer.onError(new CompositeException(Arrays.asList(e, e2)));
                     return;
                 }
                 observer.onError(e);

src/test/java/rx/internal/operators/OperatorDoOnEachTest.java

@@ -17,25 +17,19 @@
 
 import static org.junit.Assert.*;
 import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-
-import rx.Observable;
-import rx.Observer;
-import rx.Subscriber;
-import rx.exceptions.OnErrorNotImplementedException;
-import rx.functions.Action1;
-import rx.functions.Func1;
+import static org.mockito.Mockito.*;
 
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.junit.*;
+import org.mockito.*;
+
+import rx.*;
+import rx.exceptions.*;
+import rx.functions.*;
+import rx.observers.TestSubscriber;
+
 public class OperatorDoOnEachTest {
 
     @Mock
@@ -201,4 +195,28 @@ public void call(Object o) {
             System.out.println("Received exception: " + e);
         }
     }
+    
+    @Test
+    public void testOnErrorThrows() {
+        TestSubscriber<Object> ts = TestSubscriber.create();
+        
+        Observable.error(new TestException())
+        .doOnError(new Action1<Throwable>() {
+            @Override
+            public void call(Throwable e) {
+                throw new TestException();
+            }
+        }).subscribe(ts);
+        
+        ts.assertNoValues();
+        ts.assertNotCompleted();
+        ts.assertError(CompositeException.class);
+        
+        CompositeException ex = (CompositeException)ts.getOnErrorEvents().get(0);
+        
+        List<Throwable> exceptions = ex.getExceptions();
+        assertEquals(2, exceptions.size());
+        assertTrue(exceptions.get(0) instanceof TestException);
+        assertTrue(exceptions.get(1) instanceof TestException);
+    }
 }