implement Maybe.switchIfEmpty(Single) (#5582)

* implement Maybe.switchIfEmpty(Single)

* switchIfEmpty(Single) returns single; remove all changes unrelated to the PR

* add 'experimental' annotation

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

@@ -3820,6 +3820,31 @@ public final Maybe<T> switchIfEmpty(MaybeSource<? extends T> other) {
         return RxJavaPlugins.onAssembly(new MaybeSwitchIfEmpty<T>(this, other));
     }
 
+    /**
+     * Returns a Single that emits the items emitted by the source Maybe or the item of an alternate
+     * SingleSource if the current Maybe is empty.
+     * <p>
+     * <img width="640" height="445" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/switchifempty.m.png" alt="">
+     * <p/>
+     * <dl>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>{@code switchIfEmpty} does not operate by default on a particular {@link Scheduler}.</dd>
+     * </dl>
+     *
+     * @param other
+     *              the alternate SingleSource to subscribe to if the main does not emit any items
+     * @return  a Single that emits the items emitted by the source Maybe or the item of an
+     *          alternate SingleSource if the source Maybe is empty.
+     * @since 2.1.4 - experimental
+     */
+    @Experimental
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final Single<T> switchIfEmpty(SingleSource<? extends T> other) {
+        ObjectHelper.requireNonNull(other, "other is null");
+        return RxJavaPlugins.onAssembly(new MaybeSwitchIfEmptySingle<T>(this, other));
+    }
+
     /**
      * Returns a Maybe that emits the items emitted by the source Maybe until a second MaybeSource
      * emits an item.

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

@@ -0,0 +1,124 @@
+/**
+ * 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.maybe;
+
+import io.reactivex.*;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+import io.reactivex.internal.fuseable.HasUpstreamMaybeSource;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Subscribes to the other source if the main source is empty.
+ *
+ * @param <T> the value type
+ */
+public final class MaybeSwitchIfEmptySingle<T> extends Single<T> implements HasUpstreamMaybeSource<T> {
+
+    final MaybeSource<T> source;
+    final SingleSource<? extends T> other;
+
+    public MaybeSwitchIfEmptySingle(MaybeSource<T> source, SingleSource<? extends T> other) {
+        this.source = source;
+        this.other = other;
+    }
+
+    @Override
+    public MaybeSource<T> source() {
+        return source;
+    }
+
+    @Override
+    protected void subscribeActual(SingleObserver<? super T> observer) {
+        source.subscribe(new SwitchIfEmptyMaybeObserver<T>(observer, other));
+    }
+
+    static final class SwitchIfEmptyMaybeObserver<T>
+    extends AtomicReference<Disposable>
+    implements MaybeObserver<T>, Disposable {
+
+        private static final long serialVersionUID = 4603919676453758899L;
+
+        final SingleObserver<? super T> actual;
+
+        final SingleSource<? extends T> other;
+
+        SwitchIfEmptyMaybeObserver(SingleObserver<? super T> actual, SingleSource<? extends T> other) {
+            this.actual = actual;
+            this.other = other;
+        }
+
+        @Override
+        public void dispose() {
+            DisposableHelper.dispose(this);
+        }
+
+        @Override
+        public boolean isDisposed() {
+            return DisposableHelper.isDisposed(get());
+        }
+
+        @Override
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.setOnce(this, d)) {
+                actual.onSubscribe(this);
+            }
+        }
+
+        @Override
+        public void onSuccess(T value) {
+            actual.onSuccess(value);
+        }
+
+        @Override
+        public void onError(Throwable e) {
+            actual.onError(e);
+        }
+
+        @Override
+        public void onComplete() {
+            Disposable d = get();
+            if (d != DisposableHelper.DISPOSED) {
+                if (compareAndSet(d, null)) {
+                    other.subscribe(new OtherSingleObserver<T>(actual, this));
+                }
+            }
+        }
+
+        static final class OtherSingleObserver<T> implements SingleObserver<T> {
+
+            final SingleObserver<? super T> actual;
+
+            final AtomicReference<Disposable> parent;
+            OtherSingleObserver(SingleObserver<? super T> actual, AtomicReference<Disposable> parent) {
+                this.actual = actual;
+                this.parent = parent;
+            }
+            @Override
+            public void onSubscribe(Disposable d) {
+                DisposableHelper.setOnce(parent, d);
+            }
+            @Override
+            public void onSuccess(T value) {
+                actual.onSuccess(value);
+            }
+            @Override
+            public void onError(Throwable e) {
+                actual.onError(e);
+            }
+        }
+
+    }
+}

src/test/java/io/reactivex/internal/operators/maybe/MaybeSwitchIfEmptySingleTest.java

@@ -0,0 +1,108 @@
+/**
+ * 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.maybe;
+
+import io.reactivex.Maybe;
+import io.reactivex.Single;
+import io.reactivex.TestHelper;
+import io.reactivex.exceptions.TestException;
+import io.reactivex.functions.Function;
+import io.reactivex.observers.TestObserver;
+import io.reactivex.processors.PublishProcessor;
+import io.reactivex.schedulers.Schedulers;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class MaybeSwitchIfEmptySingleTest {
+
+    @Test
+    public void nonEmpty() {
+        Maybe.just(1).switchIfEmpty(Single.just(2)).test().assertResult(1);
+    }
+
+    @Test
+    public void empty() {
+        Maybe.<Integer>empty().switchIfEmpty(Single.just(2)).test().assertResult(2);
+    }
+
+    @Test
+    public void error() {
+        Maybe.<Integer>error(new TestException()).switchIfEmpty(Single.just(2))
+        .test().assertFailure(TestException.class);
+    }
+
+    @Test
+    public void errorOther() {
+        Maybe.empty().switchIfEmpty(Single.<Integer>error(new TestException()))
+        .test().assertFailure(TestException.class);
+    }
+
+    @Test
+    public void dispose() {
+        PublishProcessor<Integer> pp = PublishProcessor.create();
+
+        TestObserver<Integer> ts = pp.singleElement().switchIfEmpty(Single.just(2)).test();
+
+        assertTrue(pp.hasSubscribers());
+
+        ts.cancel();
+
+        assertFalse(pp.hasSubscribers());
+    }
+
+
+    @Test
+    public void isDisposed() {
+        PublishProcessor<Integer> pp = PublishProcessor.create();
+
+        TestHelper.checkDisposed(pp.singleElement().switchIfEmpty(Single.just(2)));
+    }
+
+    @Test
+    public void doubleOnSubscribe() {
+        TestHelper.checkDoubleOnSubscribeMaybeToSingle(new Function<Maybe<Integer>, Single<Integer>>() {
+            @Override
+            public Single<Integer> apply(Maybe<Integer> f) throws Exception {
+                return f.switchIfEmpty(Single.just(2));
+            }
+        });
+    }
+
+    @Test
+    public void emptyCancelRace() {
+        for (int i = 0; i < 500; i++) {
+            final PublishProcessor<Integer> pp = PublishProcessor.create();
+
+            final TestObserver<Integer> ts = pp.singleElement().switchIfEmpty(Single.just(2)).test();
+
+            Runnable r1 = new Runnable() {
+                @Override
+                public void run() {
+                    pp.onComplete();
+                }
+            };
+
+            Runnable r2 = new Runnable() {
+                @Override
+                public void run() {
+                    ts.cancel();
+                }
+            };
+
+            TestHelper.race(r1, r2, Schedulers.single());
+        }
+    }
+}