Merge pull request #3562 from artem-zinnatullin/single-do-on-unsubscribe

Add Single.doOnUnsubscribe()

Author: Aaron Tull

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.OperatorDoOnUnsubscribe;
 import rx.internal.operators.OperatorMap;
 import rx.internal.operators.OperatorObserveOn;
 import rx.internal.operators.OperatorOnErrorReturn;
@@ -1998,4 +2000,24 @@ public void call(SingleSubscriber<? super T> singleSubscriber) {
             }
         });
     }
+
+    /**
+     * Modifies the source {@link Single} so that it invokes the given action when it is unsubscribed from
+     * its subscribers.
+     * <p>
+     * <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/doOnUnsubscribe.png" alt="">
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code doOnUnsubscribe} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param action
+     *            the action that gets called when this {@link Single} is unsubscribed.
+     * @return the source {@link Single} modified so as to call this Action when appropriate.
+     * @see <a href="http://reactivex.io/documentation/operators/do.html">ReactiveX operators documentation: Do</a>
+     */
+    @Experimental
+    public final Single<T> doOnUnsubscribe(final Action0 action) {
+        return lift(new OperatorDoOnUnsubscribe<T>(action));
+    }
 }

src/test/java/rx/SingleTest.java

@@ -40,6 +40,7 @@
 import org.mockito.stubbing.Answer;
 import rx.Single.OnSubscribe;
 import rx.exceptions.CompositeException;
+import rx.functions.Action;
 import rx.functions.Action0;
 import rx.functions.Action1;
 import rx.functions.Func1;
@@ -828,4 +829,66 @@ public void deferShouldPassNullPointerExceptionToTheSubscriberIfSingleFactoryRet
 
         verify(singleFactory).call();
     }
+
+    @Test
+    public void doOnUnsubscribeShouldInvokeActionAfterSuccess() {
+        Action0 action = mock(Action0.class);
+
+        Single<String> single = Single
+            .just("test")
+            .doOnUnsubscribe(action);
+
+        verifyZeroInteractions(action);
+
+        TestSubscriber<String> testSubscriber = new TestSubscriber<String>();
+        single.subscribe(testSubscriber);
+
+        testSubscriber.assertValue("test");
+        testSubscriber.assertCompleted();
+
+        verify(action).call();
+    }
+
+    @Test
+    public void doOnUnsubscribeShouldInvokeActionAfterError() {
+        Action0 action = mock(Action0.class);
+
+        Single<Object> single = Single
+            .error(new RuntimeException("test"))
+            .doOnUnsubscribe(action);
+
+        verifyZeroInteractions(action);
+
+        TestSubscriber<Object> testSubscriber = new TestSubscriber<Object>();
+        single.subscribe(testSubscriber);
+
+        testSubscriber.assertError(RuntimeException.class);
+        assertEquals("test", testSubscriber.getOnErrorEvents().get(0).getMessage());
+
+        verify(action).call();
+    }
+
+    @Test
+    public void doOnUnsubscribeShouldInvokeActionAfterExplicitUnsubscription() {
+        Action0 action = mock(Action0.class);
+
+        Single<Object> single = Single
+            .create(new OnSubscribe<Object>() {
+                @Override
+                public void call(SingleSubscriber<? super Object> singleSubscriber) {
+                    // Broken Single that never ends itself (simulates long computation in one thread).
+                }
+            })
+            .doOnUnsubscribe(action);
+
+        TestSubscriber<Object> testSubscriber = new TestSubscriber<Object>();
+        Subscription subscription = single.subscribe(testSubscriber);
+
+        verifyZeroInteractions(action);
+
+        subscription.unsubscribe();
+        verify(action).call();
+        testSubscriber.assertNoValues();
+        testSubscriber.assertNoTerminalEvent();
+    }
 }