New operation Finally (issue #43)

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

@@ -40,6 +40,7 @@
 import rx.operators.OperationDefer;
 import rx.operators.OperationDematerialize;
 import rx.operators.OperationFilter;
+import rx.operators.OperationFinally;
 import rx.operators.OperationWhere;
 import rx.operators.OperationMap;
 import rx.operators.OperationMaterialize;
@@ -1182,6 +1183,18 @@ public static <T> Observable<T> concat(Observable<T>... source) {
         return _create(OperationConcat.concat(source));
     }
 
+    /**
+     * Emits the same objects as the given Observable, calling the given action
+     * when it calls <code>onComplete</code> or <code>onError</code>.
+     * @param source an observable
+     * @param action an action to be called when the source completes or errors.
+     * @return an Observable that emits the same objects, then calls the action.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh212133(v=vs.103).aspx">MSDN: Observable.Finally Method</a>
+     */
+    public static <T> Observable<T> finally0(Observable source, Action0 action) {
+        return _create(OperationFinally.finally0(source, action));
+    }
+
     /**
      * Groups the elements of an observable and selects the resulting elements by using a specified function.
      * 
@@ -2413,6 +2426,17 @@ public Observable<T> filter(Func1<T, Boolean> predicate) {
         return filter(this, predicate);
     }
 
+    /**
+     * Registers an action to be called when this observable calls
+     * <code>onComplete</code> or <code>onError</code>.
+     * @param action an action to be called when this observable completes or errors.
+     * @return an Observable that emits the same objects as this observable, then calls the action.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh212133(v=vs.103).aspx">MSDN: Observable.Finally Method</a>
+     */
+    public Observable<T> finally0(Action0 action) {
+        return _create(OperationFinally.finally0(this, action));
+    }
+
     /**
      * Filters an Observable by discarding any of its emissions that do not meet some test.
      * <p>

rxjava-core/src/main/java/rx/operators/OperationFinally.java

@@ -0,0 +1,132 @@
+/**
+ * Copyright 2013 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import rx.Observable;
+import rx.Observer;
+import rx.Subscription;
+import rx.util.AtomicObservableSubscription;
+import rx.util.functions.Action0;
+import rx.util.functions.Func1;
+
+public final class OperationFinally {
+
+    /**
+     * Call a given action when a sequence completes (with or without an
+     * exception).  The returned observable is exactly as threadsafe as the
+     * source observable; in particular, any situation allowing the source to
+     * call onComplete or onError multiple times allows the returned observable
+     * to call the final action multiple times.
+     * <p/>
+     * Note that "finally" is a Java reserved word and cannot be an identifier,
+     * so we use "finally0".
+     *
+     * @param sequence An observable sequence of elements
+     * @param action An action to be taken when the sequence is complete or throws an exception
+     * @return An observable sequence with the same elements as the input.
+     *         After the last element is consumed (and {@link Observer#onCompleted} has been called),
+     *         or after an exception is thrown (and {@link Observer#onError} has been called),
+     *         the given action will be called.
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh212133(v=vs.103).aspx">MSDN Observable.Finally method</a>
+     */
+    public static <T> Func1<Observer<T>, Subscription> finally0(final Observable<T> sequence, final Action0 action) {
+        return new Func1<Observer<T>, Subscription>() {
+            @Override
+            public Subscription call(Observer<T> observer) {
+                return new Finally<T>(sequence, action).call(observer);
+            }
+        };
+    }
+
+    private static class Finally<T> implements Func1<Observer<T>, Subscription> {
+        private final Observable<T> sequence;
+        private final Action0 finalAction;
+        private Subscription s;
+
+        Finally(final Observable<T> sequence, Action0 finalAction) {
+            this.sequence = sequence;
+            this.finalAction = finalAction;
+        }
+
+        private final AtomicObservableSubscription Subscription = new AtomicObservableSubscription();
+
+        private final Subscription actualSubscription = new Subscription() {
+            @Override
+            public void unsubscribe() {
+                if (null != s)
+                    s.unsubscribe();
+            }
+        };
+
+        public Subscription call(Observer<T> observer) {
+            s = sequence.subscribe(new FinallyObserver(observer));
+            return Subscription.wrap(actualSubscription);
+        }
+
+        private class FinallyObserver implements Observer<T> {
+            private final Observer<T> observer;
+
+            FinallyObserver(Observer<T> observer) {
+                this.observer = observer;
+            }
+
+            @Override
+            public void onCompleted() {
+                observer.onCompleted();
+                finalAction.call();
+            }
+
+            @Override
+            public void onError(Exception e) {
+                observer.onError(e);
+                finalAction.call();
+            }
+
+            @Override
+            public void onNext(T args) {
+                observer.onNext(args);
+            }
+        }
+    }
+
+    public static class UnitTest {
+        private Action0 aAction0;
+        private Observer<String> aObserver;
+        @Before
+        public void before() {
+            aAction0 = mock(Action0.class);
+            aObserver = mock(Observer.class);
+        }
+        private void checkActionCalled(Observable<String> input) {
+            Observable.create(finally0(input, aAction0)).subscribe(aObserver);
+            verify(aAction0, times(1)).call();
+        }
+        @Test
+        public void testFinallyCalledOnComplete() {
+            checkActionCalled(Observable.toObservable(new String[] {"1", "2", "3"}));
+        }
+        @Test
+        public void testFinallyCalledOnError() {
+            checkActionCalled(Observable.<String>error(new RuntimeException("expected")));
+        }
+    }
+}