Merge pull request #918 from benjchristensen/doOnTerminate

Operator: doOnTerminate

Author: benjchristensen

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

@@ -4293,17 +4293,17 @@ public final Observable<T> doOnEach(final Action1<Notification<? super T>> onNot
         Observer<T> observer = new Observer<T>() {
             @Override
             public final void onCompleted() {
-                onNotification.call(new Notification<T>());
+                onNotification.call(Notification.createOnCompleted());
             }
 
             @Override
             public final void onError(Throwable e) {
-                onNotification.call(new Notification<T>(e));
+                onNotification.call(Notification.createOnError(e));
             }
 
             @Override
             public final void onNext(T v) {
-                onNotification.call(new Notification<T>(v));
+                onNotification.call(Notification.createOnNext(v));
             }
 
         };
@@ -4387,6 +4387,39 @@ public final void onNext(T args) {
 
         return lift(new OperatorDoOnEach<T>(observer));
     }
+    
+    /**
+     * Modifies an Observable so that it invokes an action when it calls {@code onCompleted} or {@code onError} <p>
+     * <img width="640" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/doOnCompleted.png">
+     * <p>
+     * This differs from {@code finallyDo} in that this happens BEFORE onCompleted/onError are emitted.
+     * 
+     * @param onTerminate
+     *            the action to invoke when the source Observable calls {@code onCompleted} or {@code onError}
+     * @return the source Observable with the side-effecting behavior applied
+     * @see <a href="https://github.com/Netflix/RxJava/wiki/Observable-Utility-Operators#wiki-dooncompleted">RxJava Wiki: doOnCompleted()</a>
+     * @see <a href="http://msdn.microsoft.com/en-us/library/hh229804.aspx">MSDN: Observable.Do</a>
+     */
+    public final Observable<T> doOnTerminate(final Action0 onTerminate) {
+        Observer<T> observer = new Observer<T>() {
+            @Override
+            public final void onCompleted() {
+                onTerminate.call();
+            }
+
+            @Override
+            public final void onError(Throwable e) {
+                onTerminate.call();
+            }
+
+            @Override
+            public final void onNext(T args) {
+            }
+
+        };
+
+        return lift(new OperatorDoOnEach<T>(observer));
+    }
 
     /**
      * Returns an Observable that emits the single item at a specified index in a sequence of emissions from a