2.x: Add Flowable.switchMapCompletable{DelayError} operator

Author: akarnokd

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

@@ -13951,6 +13951,97 @@ public final <R> Flowable<R> switchMap(Function<? super T, ? extends Publisher<?
         return switchMap0(mapper, bufferSize, false);
     }
 
+    /**
+     * Maps the upstream values into {@link CompletableSource}s, subscribes to the newer one while
+     * disposing the subscription to the previous {@code CompletableSource}, thus keeping at most one
+     * active {@code CompletableSource} running.
+     * <p>
+     * <img width="640" height="521" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/switchMapCompletable.f.png" alt="">
+     * <p>
+     * Since a {@code CompletableSource} doesn't produce any items, the resulting reactive type of
+     * this operator is a {@link Completable} that can only indicate successful completion or
+     * a failure in any of the inner {@code CompletableSource}s or the failure of the current
+     * {@link Flowable}.
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>The operator consumes the current {@link Flowable} in an unbounded manner and otherwise
+     *  does not have backpressure in its return type because no items are ever produced.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code switchMapCompletable} does not operate by default on a particular {@link Scheduler}.</dd>
+     *  <dt><b>Error handling:</b></dt>
+     *  <dd>If either this {@code Flowable} or the active {@code CompletableSource} signals an {@code onError},
+     *  the resulting {@code Completable} is terminated immediately with that {@code Throwable}.
+     *  Use the {@link #switchMapCompletableDelayError(Function)} to delay such inner failures until
+     *  every inner {@code CompletableSource}s and the main {@code Flowable} terminates in some fashion.
+     *  If they fail concurrently, the operator may combine the {@code Throwable}s into a
+     *  {@link io.reactivex.exceptions.CompositeException CompositeException}
+     *  and signal it to the downstream instead. If any inactivated (switched out) {@code CompletableSource}
+     *  signals an {@code onError} late, the {@code Throwable}s will be signalled to the global error handler via
+     *  {@link RxJavaPlugins#onError(Throwable)} method as {@code UndeliverableException} errors.
+     *  </dd>
+     * </dl>
+     * @param mapper the function called with each upstream item and should return a
+     *               {@link CompletableSource} to be subscribed to and awaited for
+     *               (non blockingly) for its terminal event
+     * @return the new Completable instance
+     * @since 2.1.11 - experimental
+     * @see #switchMapCompletableDelayError(Function)
+     */
+    @CheckReturnValue
+    @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @Experimental
+    public final Completable switchMapCompletable(@NonNull Function<? super T, ? extends CompletableSource> mapper) {
+        ObjectHelper.requireNonNull(mapper, "mapper is null");
+        return RxJavaPlugins.onAssembly(new FlowableSwitchMapCompletable<T>(this, mapper, false));
+    }
+
+    /**
+     * Maps the upstream values into {@link CompletableSource}s, subscribes to the newer one while
+     * disposing the subscription to the previous {@code CompletableSource}, thus keeping at most one
+     * active {@code CompletableSource} running and delaying any main or inner errors until all
+     * of them terminate.
+     * <p>
+     * <img width="640" height="453" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/switchMapCompletableDelayError.f.png" alt="">
+     * <p>
+     * Since a {@code CompletableSource} doesn't produce any items, the resulting reactive type of
+     * this operator is a {@link Completable} that can only indicate successful completion or
+     * a failure in any of the inner {@code CompletableSource}s or the failure of the current
+     * {@link Flowable}.
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>The operator consumes the current {@link Flowable} in an unbounded manner and otherwise
+     *  does not have backpressure in its return type because no items are ever produced.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code switchMapCompletableDelayError} does not operate by default on a particular {@link Scheduler}.</dd>
+     *  <dt><b>Error handling:</b></dt>
+     *  <dd>Errors of this {@code Flowable} and all the {@code CompletableSource}s, who had the chance
+     *  to run to their completion, are delayed until
+     *  all of the terminate in some fashion. At this point, if there was only one failure, the respective
+     *  {@code Throwable} is emitted to the dowstream. It there were more than one failures, the
+     *  operator combines all {@code Throwable}s into a {@link io.reactivex.exceptions.CompositeException CompositeException}
+     *  and signals that to the downstream.
+     *  If any inactivated (switched out) {@code CompletableSource}
+     *  signals an {@code onError} late, the {@code Throwable}s will be signalled to the global error handler via
+     *  {@link RxJavaPlugins#onError(Throwable)} method as {@code UndeliverableException} errors.
+     *  </dd>
+     * </dl>
+     * @param mapper the function called with each upstream item and should return a
+     *               {@link CompletableSource} to be subscribed to and awaited for
+     *               (non blockingly) for its terminal event
+     * @return the new Completable instance
+     * @since 2.1.11 - experimental
+     * @see #switchMapCompletableDelayError(Function)
+     */
+    @CheckReturnValue
+    @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
+    @SchedulerSupport(SchedulerSupport.NONE)
+    @Experimental
+    public final Completable switchMapCompletableDelayError(@NonNull Function<? super T, ? extends CompletableSource> mapper) {
+        ObjectHelper.requireNonNull(mapper, "mapper is null");
+        return RxJavaPlugins.onAssembly(new FlowableSwitchMapCompletable<T>(this, mapper, true));
+    }
+
     /**
      * Returns a new Publisher by applying a function that you supply to each item emitted by the source
      * Publisher that returns a Publisher, and then emitting the items emitted by the most recently emitted

src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchMapCompletable.java

@@ -0,0 +1,239 @@
+/**
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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.flowable;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.reactivestreams.Subscription;
+
+import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.exceptions.Exceptions;
+import io.reactivex.functions.Function;
+import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.functions.ObjectHelper;
+import io.reactivex.internal.subscriptions.SubscriptionHelper;
+import io.reactivex.internal.util.*;
+import io.reactivex.plugins.RxJavaPlugins;
+
+/**
+ * Maps the upstream values into {@link CompletableSource}s, subscribes to the newer one while
+ * disposing the subscription to the previous {@code CompletableSource}, thus keeping at most one
+ * active {@code CompletableSource} running.
+ *
+ * @param <T> the upstream value type
+ * @since 2.1.11 - experimental
+ */
+@Experimental
+public final class FlowableSwitchMapCompletable<T> extends Completable {
+
+    final Flowable<T> source;
+
+    final Function<? super T, ? extends CompletableSource> mapper;
+
+    final boolean delayErrors;
+
+    public FlowableSwitchMapCompletable(Flowable<T> source,
+            Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors) {
+        this.source = source;
+        this.mapper = mapper;
+        this.delayErrors = delayErrors;
+    }
+
+    @Override
+    protected void subscribeActual(CompletableObserver s) {
+        source.subscribe(new SwitchMapCompletableObserver<T>(s, mapper, delayErrors));
+    }
+
+    static final class SwitchMapCompletableObserver<T> implements FlowableSubscriber<T>, Disposable {
+
+        final CompletableObserver downstream;
+
+        final Function<? super T, ? extends CompletableSource> mapper;
+
+        final boolean delayErrors;
+
+        final AtomicThrowable errors;
+
+        final AtomicReference<SwitchMapInnerObserver> inner;
+
+        static final SwitchMapInnerObserver INNER_DISPOSED = new SwitchMapInnerObserver(null);
+
+        volatile boolean done;
+
+        Subscription upstream;
+
+        SwitchMapCompletableObserver(CompletableObserver downstream,
+                Function<? super T, ? extends CompletableSource> mapper, boolean delayErrors) {
+            this.downstream = downstream;
+            this.mapper = mapper;
+            this.delayErrors = delayErrors;
+            this.errors = new AtomicThrowable();
+            this.inner = new AtomicReference<SwitchMapInnerObserver>();
+        }
+
+        @Override
+        public void onSubscribe(Subscription s) {
+            if (SubscriptionHelper.validate(upstream, s)) {
+                this.upstream = s;
+                downstream.onSubscribe(this);
+                s.request(Long.MAX_VALUE);
+            }
+        }
+
+        @Override
+        public void onNext(T t) {
+            CompletableSource c;
+
+            try {
+                c = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null CompletableSource");
+            } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
+                upstream.cancel();
+                onError(ex);
+                return;
+            }
+
+            SwitchMapInnerObserver o = new SwitchMapInnerObserver(this);
+
+            for (;;) {
+                SwitchMapInnerObserver current = inner.get();
+                if (current == INNER_DISPOSED) {
+                    break;
+                }
+                if (inner.compareAndSet(current, o)) {
+                    if (current != null) {
+                        current.dispose();
+                    }
+                    c.subscribe(o);
+                    break;
+                }
+            }
+        }
+
+        @Override
+        public void onError(Throwable t) {
+            if (errors.addThrowable(t)) {
+                if (delayErrors) {
+                    onComplete();
+                } else {
+                    disposeInner();
+                    Throwable ex = errors.terminate();
+                    if (ex != ExceptionHelper.TERMINATED) {
+                        downstream.onError(ex);
+                    }
+                }
+            } else {
+                RxJavaPlugins.onError(t);
+            }
+        }
+
+        @Override
+        public void onComplete() {
+            done = true;
+            if (inner.get() == null) {
+                Throwable ex = errors.terminate();
+                if (ex == null) {
+                    downstream.onComplete();
+                } else {
+                    downstream.onError(ex);
+                }
+            }
+        }
+
+        void disposeInner() {
+            SwitchMapInnerObserver o = inner.getAndSet(INNER_DISPOSED);
+            if (o != null && o != INNER_DISPOSED) {
+                o.dispose();
+            }
+        }
+
+        @Override
+        public void dispose() {
+            upstream.cancel();
+            disposeInner();
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return inner.get() == INNER_DISPOSED;
+        }
+
+        void innerError(SwitchMapInnerObserver sender, Throwable error) {
+            if (inner.compareAndSet(sender, null)) {
+                if (errors.addThrowable(error)) {
+                    if (delayErrors) {
+                        if (done) {
+                            Throwable ex = errors.terminate();
+                            downstream.onError(ex);
+                        }
+                    } else {
+                        dispose();
+                        Throwable ex = errors.terminate();
+                        if (ex != ExceptionHelper.TERMINATED) {
+                            downstream.onError(ex);
+                        }
+                    }
+                    return;
+                }
+            }
+            RxJavaPlugins.onError(error);
+        }
+
+        void innerComplete(SwitchMapInnerObserver sender) {
+            if (inner.compareAndSet(sender, null)) {
+                if (done) {
+                    Throwable ex = errors.terminate();
+                    if (ex == null) {
+                        downstream.onComplete();
+                    } else {
+                        downstream.onError(ex);
+                    }
+                }
+            }
+        }
+
+        static final class SwitchMapInnerObserver extends AtomicReference<Disposable>
+        implements CompletableObserver {
+
+            private static final long serialVersionUID = -8003404460084760287L;
+
+            final SwitchMapCompletableObserver<?> parent;
+
+            SwitchMapInnerObserver(SwitchMapCompletableObserver<?> parent) {
+                this.parent = parent;
+            }
+
+            @Override
+            public void onSubscribe(Disposable d) {
+                DisposableHelper.setOnce(this, d);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                parent.innerError(this, e);
+            }
+
+            @Override
+            public void onComplete() {
+                parent.innerComplete(this);
+            }
+
+            void dispose() {
+                DisposableHelper.dispose(this);
+            }
+        }
+    }
+}

src/test/java/io/reactivex/InternalWrongNaming.java

@@ -185,7 +185,10 @@ public void flowableNoObserver() throws Exception {
                 "FlowableConcatWithCompletable",
                 "FlowableMergeWithSingle",
                 "FlowableMergeWithMaybe",
-                "FlowableMergeWithCompletable"
+                "FlowableMergeWithCompletable",
+                "FlowableSwitchMapCompletable",
+                "FlowableSwitchMapSingle",
+                "FlowableSwitchMapMaybe"
         );
     }
 }