2.x: add doFinally to the rest of the reactive base classes

src/main/java/io/reactivex/Completable.java

@@ -25,8 +25,7 @@
 import io.reactivex.internal.observers.*;
 import io.reactivex.internal.operators.completable.*;
 import io.reactivex.internal.operators.flowable.FlowableDelaySubscriptionOther;
-import io.reactivex.internal.operators.maybe.MaybeFromCompletable;
-import io.reactivex.internal.operators.maybe.MaybeDelayWithCompletable;
+import io.reactivex.internal.operators.maybe.*;
 import io.reactivex.internal.operators.observable.ObservableDelaySubscriptionOther;
 import io.reactivex.internal.operators.single.SingleDelayWithCompletable;
 import io.reactivex.internal.util.ExceptionHelper;
@@ -1171,6 +1170,27 @@ public final Completable doAfterTerminate(final Action onAfterTerminate) {
                 onAfterTerminate,
                 Functions.EMPTY_ACTION);
     }
+    /**
+     * Calls the specified action after this Completable signals onError or onComplete or gets disposed by
+     * the downstream.
+     * <p>In case of a race between a terminal event and a dispose call, the provided {@code onFinally} action
+     * is executed once per subscription.
+     * <p>Note that the {@code onFinally} action is shared between subscriptions and as such
+     * should be thread-safe.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code doFinally} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @param onFinally the action called when this Completable terminates or gets cancelled
+     * @return the new Completable instance
+     * @since 2.0.1 - experimental
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @Experimental
+    public final Completable doFinally(Action onFinally) {
+        ObjectHelper.requireNonNull(onFinally, "onFinally is null");
+        return RxJavaPlugins.onAssembly(new CompletableDoFinally(this, onFinally));
+    }
 
     /**
      * <strong>Advanced use without safeguards:</strong> lifts a CompletableOperator

src/main/java/io/reactivex/Flowable.java

@@ -7328,7 +7328,7 @@ public final Flowable<T> distinctUntilChanged(BiPredicate<? super T, ? super T>
      * Calls the specified action after this Flowable signals onError or onCompleted or gets cancelled by
      * the downstream.
      * <p>In case of a race between a terminal event and a cancellation, the provided {@code onFinally} action
-     * is executed at once per subscription.
+     * is executed once per subscription.
      * <p>Note that the {@code onFinally} action is shared between subscriptions and as such
      * should be thread-safe.
      * <dl>

src/main/java/io/reactivex/Maybe.java

@@ -2302,6 +2302,28 @@ public final Maybe<T> doAfterTerminate(Action onAfterTerminate) {
         ));
     }
 
+    /**
+     * Calls the specified action after this Maybe signals onSuccess, onError or onComplete or gets disposed by
+     * the downstream.
+     * <p>In case of a race between a terminal event and a dispose call, the provided {@code onFinally} action
+     * is executed once per subscription.
+     * <p>Note that the {@code onFinally} action is shared between subscriptions and as such
+     * should be thread-safe.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code doFinally} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @param onFinally the action called when this Maybe terminates or gets cancelled
+     * @return the new Maybe instance
+     * @since 2.0.1 - experimental
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @Experimental
+    public final Maybe<T> doFinally(Action onFinally) {
+        ObjectHelper.requireNonNull(onFinally, "onFinally is null");
+        return RxJavaPlugins.onAssembly(new MaybeDoFinally<T>(this, onFinally));
+    }
+
     /**
      * Calls the shared runnable if a MaybeObserver subscribed to the current Maybe
      * disposes the common Disposable it received via onSubscribe.

src/main/java/io/reactivex/Observable.java

@@ -16,7 +16,6 @@
 import java.util.*;
 import java.util.concurrent.*;
 
-import io.reactivex.internal.operators.flowable.FlowableOnBackpressureError;
 import org.reactivestreams.Publisher;
 
 import io.reactivex.annotations.*;
@@ -26,7 +25,7 @@
 import io.reactivex.internal.functions.*;
 import io.reactivex.internal.fuseable.ScalarCallable;
 import io.reactivex.internal.observers.*;
-import io.reactivex.internal.operators.flowable.FlowableFromObservable;
+import io.reactivex.internal.operators.flowable.*;
 import io.reactivex.internal.operators.observable.*;
 import io.reactivex.internal.util.*;
 import io.reactivex.observables.*;
@@ -6440,6 +6439,30 @@ public final Observable<T> doAfterTerminate(Action onFinally) {
         return doOnEach(Functions.emptyConsumer(), Functions.emptyConsumer(), Functions.EMPTY_ACTION, onFinally);
     }
 
+    /**
+     * Calls the specified action after this Observable signals onError or onCompleted or gets disposed by
+     * the downstream.
+     * <p>In case of a race between a terminal event and a dispose call, the provided {@code onFinally} action
+     * is executed once per subscription.
+     * <p>Note that the {@code onFinally} action is shared between subscriptions and as such
+     * should be thread-safe.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code doFinally} does not operate by default on a particular {@link Scheduler}.</dd>
+     *  <td><b>Operator-fusion:</b></dt>
+     *  <dd>This operator supports boundary-limited synchronous or asynchronous queue-fusion.</dd>
+     * </dl>
+     * @param onFinally the action called when this Observable terminates or gets cancelled
+     * @return the new Observable instance
+     * @since 2.0.1 - experimental
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @Experimental
+    public final Observable<T> doFinally(Action onFinally) {
+        ObjectHelper.requireNonNull(onFinally, "onFinally is null");
+        return RxJavaPlugins.onAssembly(new ObservableDoFinally<T>(this, onFinally));
+    }
+
     /**
      * Calls the unsubscribe {@code Action} if the downstream disposes the sequence.
      * <p>

src/main/java/io/reactivex/Single.java

@@ -1716,6 +1716,28 @@ public final <U> Single<T> delaySubscription(long time, TimeUnit unit, Scheduler
         return delaySubscription(Observable.timer(time, unit, scheduler));
     }
 
+    /**
+     * Calls the specified action after this Single signals onSuccess or onError or gets disposed by
+     * the downstream.
+     * <p>In case of a race between a terminal event and a dispose call, the provided {@code onFinally} action
+     * is executed once per subscription.
+     * <p>Note that the {@code onFinally} action is shared between subscriptions and as such
+     * should be thread-safe.
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code doFinally} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     * @param onFinally the action called when this Single terminates or gets cancelled
+     * @return the new Single instance
+     * @since 2.0.1 - experimental
+     */
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @Experimental
+    public final Single<T> doFinally(Action onFinally) {
+        ObjectHelper.requireNonNull(onFinally, "onFinally is null");
+        return RxJavaPlugins.onAssembly(new SingleDoFinally<T>(this, onFinally));
+    }
+
     /**
      * Calls the shared consumer with the Disposable sent through the onSubscribe for each
      * SingleObserver that subscribes to the current Single.

src/main/java/io/reactivex/internal/operators/completable/CompletableDoFinally.java

@@ -0,0 +1,106 @@
+/**
+ * Copyright 2016 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 io.reactivex.internal.operators.completable;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.exceptions.Exceptions;
+import io.reactivex.functions.Action;
+import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.plugins.RxJavaPlugins;
+
+/**
+ * Execute an action after an onError, onComplete or a dispose event.
+ *
+ * @since 2.0.1 - experimental
+ */
+@Experimental
+public final class CompletableDoFinally extends Completable {
+
+    final CompletableSource source;
+
+    final Action onFinally;
+
+    public CompletableDoFinally(CompletableSource source, Action onFinally) {
+        this.source = source;
+        this.onFinally = onFinally;
+    }
+
+    @Override
+    protected void subscribeActual(CompletableObserver s) {
+        source.subscribe(new DoFinallyObserver(s, onFinally));
+    }
+
+    static final class DoFinallyObserver extends AtomicInteger implements CompletableObserver, Disposable {
+
+        private static final long serialVersionUID = 4109457741734051389L;
+
+        final CompletableObserver actual;
+
+        final Action onFinally;
+
+        Disposable d;
+
+        DoFinallyObserver(CompletableObserver actual, Action onFinally) {
+            this.actual = actual;
+            this.onFinally = onFinally;
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.validate(this.d, d)) {
+                this.d = d;
+
+                actual.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            actual.onError(t);
+            runFinally();
+        }
+
+        @Override
+        public void onComplete() {
+            actual.onComplete();
+            runFinally();
+        }
+
+        @Override
+        public void dispose() {
+            d.dispose();
+            runFinally();
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return d.isDisposed();
+        }
+
+        void runFinally() {
+            if (compareAndSet(0, 1)) {
+                try {
+                    onFinally.run();
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    RxJavaPlugins.onError(ex);
+                }
+            }
+        }
+    }
+}

src/main/java/io/reactivex/internal/operators/maybe/MaybeDoFinally.java

@@ -0,0 +1,111 @@
+/**
+ * Copyright 2016 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 io.reactivex.internal.operators.maybe;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.exceptions.Exceptions;
+import io.reactivex.functions.Action;
+import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.plugins.RxJavaPlugins;
+
+/**
+ * Execute an action after an onSuccess, onError, onComplete or a dispose event.
+ *
+ * @param <T> the value type
+ * @since 2.0.1 - experimental
+ */
+@Experimental
+public final class MaybeDoFinally<T> extends AbstractMaybeWithUpstream<T, T> {
+
+    final Action onFinally;
+
+    public MaybeDoFinally(MaybeSource<T> source, Action onFinally) {
+        super(source);
+        this.onFinally = onFinally;
+    }
+
+    @Override
+    protected void subscribeActual(MaybeObserver<? super T> s) {
+        source.subscribe(new DoFinallyObserver<T>(s, onFinally));
+    }
+
+    static final class DoFinallyObserver<T> extends AtomicInteger implements MaybeObserver<T>, Disposable {
+
+        private static final long serialVersionUID = 4109457741734051389L;
+
+        final MaybeObserver<? super T> actual;
+
+        final Action onFinally;
+
+        Disposable d;
+
+        DoFinallyObserver(MaybeObserver<? super T> actual, Action onFinally) {
+            this.actual = actual;
+            this.onFinally = onFinally;
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.validate(this.d, d)) {
+                this.d = d;
+
+                actual.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onSuccess(T t) {
+            actual.onSuccess(t);
+            runFinally();
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            actual.onError(t);
+            runFinally();
+        }
+
+        @Override
+        public void onComplete() {
+            actual.onComplete();
+            runFinally();
+        }
+
+        @Override
+        public void dispose() {
+            d.dispose();
+            runFinally();
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return d.isDisposed();
+        }
+
+        void runFinally() {
+            if (compareAndSet(0, 1)) {
+                try {
+                    onFinally.run();
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    RxJavaPlugins.onError(ex);
+                }
+            }
+        }
+    }
+}