Add Single.doOnError()

src/main/java/rx/Single.java

@@ -31,6 +31,7 @@
 import rx.functions.Func8;
 import rx.functions.Func9;
 import rx.internal.operators.OnSubscribeToObservableFuture;
+import rx.internal.operators.OperatorDoOnEach;
 import rx.internal.operators.OperatorMap;
 import rx.internal.operators.OperatorObserveOn;
 import rx.internal.operators.OperatorOnErrorReturn;
@@ -1789,4 +1790,40 @@ public final <T2, R> Single<R> zipWith(Single<? extends T2> other, Func2<? super
         return zip(this, other, zipFunction);
     }
 
+    /**
+     * Modifies the source {@link Single} 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>
+     *  <dd>{@code doOnError} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param onError
+     *            the action to invoke if the source {@link Single} calls {@code onError}
+     * @return the source {@link Single} with the side-effecting behavior applied
+     * @see <a href="http://reactivex.io/documentation/operators/do.html">ReactiveX operators documentation: Do</a>
+     */
+    @Experimental
+    public final Single<T> doOnError(final Action1<Throwable> onError) {
+        Observer<T> observer = new Observer<T>() {
+            @Override
+            public void onCompleted() {
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                onError.call(e);
+            }
+
+            @Override
+            public void onNext(T t) {
+            }
+        };
+
+        return lift(new OperatorDoOnEach<T>(observer));
+    }
 }

src/test/java/rx/SingleTest.java

@@ -13,8 +13,13 @@
 package rx;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
 
 import java.util.Arrays;
 import java.util.concurrent.CountDownLatch;
@@ -26,7 +31,9 @@
 import org.junit.Test;
 
 import rx.Single.OnSubscribe;
+import rx.exceptions.CompositeException;
 import rx.functions.Action0;
+import rx.functions.Action1;
 import rx.functions.Func1;
 import rx.functions.Func2;
 import rx.observers.TestSubscriber;
@@ -461,4 +468,66 @@ public void testToObservable() {
     	ts.assertValue("a");
     	ts.assertCompleted();
     }
+
+    @Test
+    public void doOnErrorShouldNotCallActionIfNoErrorHasOccurred() {
+        Action1<Throwable> action = mock(Action1.class);
+
+        TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
+
+        Single
+                .just("value")
+                .doOnError(action)
+                .subscribe(testSubscriber);
+
+        testSubscriber.assertValue("value");
+        testSubscriber.assertNoErrors();
+
+        verifyZeroInteractions(action);
+    }
+
+    @Test
+    public void doOnErrorShouldCallActionIfErrorHasOccurred() {
+        Action1<Throwable> action = mock(Action1.class);
+
+        TestSubscriber<Object> testSubscriber = new TestSubscriber<Object>();
+
+        Throwable error = new IllegalStateException();
+
+        Single
+                .error(error)
+                .doOnError(action)
+                .subscribe(testSubscriber);
+
+        testSubscriber.assertNoValues();
+        testSubscriber.assertError(error);
+
+        verify(action).call(error);
+    }
+
+    @Test
+    public void doOnErrorShouldThrowCompositeExceptionIfOnErrorActionThrows() {
+        Action1<Throwable> action = mock(Action1.class);
+
+
+        Throwable error = new RuntimeException();
+        Throwable exceptionFromOnErrorAction = new IllegalStateException();
+        doThrow(exceptionFromOnErrorAction).when(action).call(error);
+
+        TestSubscriber<Object> testSubscriber = new TestSubscriber<Object>();
+
+        Single
+                .error(error)
+                .doOnError(action)
+                .subscribe(testSubscriber);
+
+        testSubscriber.assertNoValues();
+        CompositeException compositeException = (CompositeException) testSubscriber.getOnErrorEvents().get(0);
+
+        assertEquals(2, compositeException.getExceptions().size());
+        assertSame(error, compositeException.getExceptions().get(0));
+        assertSame(exceptionFromOnErrorAction, compositeException.getExceptions().get(1));
+
+        verify(action).call(error);
+    }
 }