Add Single.doAfterTerminate()

Author: artem-zinnatullin

src/main/java/rx/Single.java

@@ -21,6 +21,7 @@
 import rx.annotations.Experimental;
 import rx.exceptions.Exceptions;
 import rx.exceptions.OnErrorNotImplementedException;
+import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.functions.Func1;
 import rx.functions.Func2;
@@ -34,6 +35,7 @@
 import rx.internal.operators.OnSubscribeToObservableFuture;
 import rx.internal.operators.OperatorDelay;
 import rx.internal.operators.OperatorDoOnEach;
+import rx.internal.operators.OperatorFinally;
 import rx.internal.operators.OperatorMap;
 import rx.internal.operators.OperatorObserveOn;
 import rx.internal.operators.OperatorOnErrorReturn;
@@ -1998,4 +2000,25 @@ public void call(SingleSubscriber<? super T> singleSubscriber) {
             }
         });
     }
+
+    /**
+     * Registers an {@link Action0} to be called when this {@link Single} invokes either
+     * {@link SingleSubscriber#onSuccess(Object)}  onSuccess} or {@link SingleSubscriber#onError onError}.
+     * <p>
+     * <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/finallyDo.png" alt="">
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code doAfterTerminate} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param action
+     *            an {@link Action0} to be invoked when the source {@link Single} finishes.
+     * @return a {@link Single} that emits the same item or error as the source {@link Single}, then invokes the
+     *         {@link Action0}
+     * @see <a href="http://reactivex.io/documentation/operators/do.html">ReactiveX operators documentation: Do</a>
+     */
+    @Experimental
+    public final Single<T> doAfterTerminate(Action0 action) {
+        return lift(new OperatorFinally<T>(action));
+    }
 }

src/test/java/rx/SingleTest.java

@@ -50,7 +50,6 @@
 import rx.schedulers.Schedulers;
 import rx.subscriptions.Subscriptions;
 
-
 public class SingleTest {
 
     @Test
@@ -828,4 +827,55 @@ public void deferShouldPassNullPointerExceptionToTheSubscriberIfSingleFactoryRet
 
         verify(singleFactory).call();
     }
+
+    @Test
+    public void doAfterTerminateActionShouldBeInvokedAfterOnSuccess() {
+        Action0 action = mock(Action0.class);
+
+        TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
+
+        Single
+                .just("value")
+                .doAfterTerminate(action)
+                .subscribe(testSubscriber);
+
+        testSubscriber.assertValue("value");
+        testSubscriber.assertNoErrors();
+
+        verify(action).call();
+    }
+
+    @Test
+    public void doAfterTerminateActionShouldBeInvokedAfterOnError() {
+        Action0 action = mock(Action0.class);
+
+        TestSubscriber<Object> testSubscriber = new TestSubscriber<Object>();
+
+        Throwable error = new IllegalStateException();
+
+        Single
+                .error(error)
+                .doAfterTerminate(action)
+                .subscribe(testSubscriber);
+
+        testSubscriber.assertNoValues();
+        testSubscriber.assertError(error);
+
+        verify(action).call();
+    }
+
+    @Test
+    public void doAfterTerminateActionShouldNotBeInvokedUntilSubscriberSubscribes() {
+        Action0 action = mock(Action0.class);
+
+        Single
+                .just("value")
+                .doAfterTerminate(action);
+
+        Single
+                .error(new IllegalStateException())
+                .doAfterTerminate(action);
+
+        verifyZeroInteractions(action);
+    }
 }