Merge pull request #222 from joshgord/pull-issue-43-merge

Pull issue 43 merge

Author: mattrjacobs

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

@@ -42,6 +42,7 @@
 import rx.operators.OperationDefer;
 import rx.operators.OperationDematerialize;
 import rx.operators.OperationFilter;
+import rx.operators.OperationFinally;
 import rx.operators.OperationMap;
 import rx.operators.OperationMaterialize;
 import rx.operators.OperationMerge;
@@ -1221,6 +1222,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> finallyDo(Observable source, Action0 action) {
+        return create(OperationFinally.finallyDo(source, action));
+    }
+
     /**
      * Groups the elements of an observable and selects the resulting elements by using a specified function.
      * 
@@ -2491,6 +2504,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> finallyDo(Action0 action) {
+        return create(OperationFinally.finallyDo(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,124 @@
+/**
+ * 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.
+     * <p/>
+     * Note that "finally" is a Java reserved word and cannot be an identifier,
+     * so we use "finallyDo".
+     *
+     * @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> finallyDo(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;
+
+        Finally(final Observable<T> sequence, Action0 finalAction) {
+            this.sequence = sequence;
+            this.finalAction = finalAction;
+        }
+
+        public Subscription call(Observer<T> observer) {
+            return sequence.subscribe(new FinallyObserver(observer));
+        }
+
+        private class FinallyObserver implements Observer<T> {
+            private final Observer<T> observer;
+
+            FinallyObserver(Observer<T> observer) {
+                this.observer = observer;
+            }
+
+            @Override
+            public void onCompleted() {
+                try {
+                    observer.onCompleted();
+                } finally {
+                    finalAction.call();
+                }
+            }
+
+            @Override
+            public void onError(Exception e) {
+                try {
+                    observer.onError(e);
+                } finally {
+                    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(finallyDo(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")));
+        }
+    }
+}