implement Maybe.switchIfEmpty(Single)

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

@@ -2213,7 +2213,7 @@ public final Single<Long> count() {
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Maybe<T> defaultIfEmpty(T defaultItem) {
         ObjectHelper.requireNonNull(defaultItem, "item is null");
-        return switchIfEmpty(just(defaultItem));
+        return switchIfEmpty(Single.just(defaultItem));
     }
 
 
@@ -3817,7 +3817,30 @@ public final <E extends MaybeObserver<? super T>> E subscribeWith(E observer) {
     @SchedulerSupport(SchedulerSupport.NONE)
     public final Maybe<T> switchIfEmpty(MaybeSource<? extends T> other) {
         ObjectHelper.requireNonNull(other, "other is null");
-        return RxJavaPlugins.onAssembly(new MaybeSwitchIfEmpty<T>(this, other));
+        return RxJavaPlugins.onAssembly(new MaybeSwitchIfEmptyToMaybe<T>(this, other));
+    }
+
+    /**
+     * Returns a Maybe 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 Maybe that emits the items emitted by the source Maybe or the item of an
+     *          alternate SingleSource if the source Maybe is empty.
+     */
+    @CheckReturnValue
+    @SchedulerSupport(SchedulerSupport.NONE)
+    public final Maybe<T> switchIfEmpty(SingleSource<? extends T> other) {
+        ObjectHelper.requireNonNull(other, "other is null");
+        return RxJavaPlugins.onAssembly(new MaybeSwitchIfEmptyToSingle<T>(this, other));
     }
 
     /**

src/main/java/io/reactivex/internal/operators/maybe/MaybeSwitchIfEmpty.java → src/main/java/io/reactivex/internal/operators/maybe/MaybeSwitchIfEmptyToMaybe.java

@@ -24,11 +24,11 @@
  *
  * @param <T> the value type
  */
-public final class MaybeSwitchIfEmpty<T> extends AbstractMaybeWithUpstream<T, T> {
+public final class MaybeSwitchIfEmptyToMaybe<T> extends AbstractMaybeWithUpstream<T, T> {
 
     final MaybeSource<? extends T> other;
 
-    public MaybeSwitchIfEmpty(MaybeSource<T> source, MaybeSource<? extends T> other) {
+    public MaybeSwitchIfEmptyToMaybe(MaybeSource<T> source, MaybeSource<? extends T> other) {
         super(source);
         this.other = other;
     }

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

@@ -0,0 +1,120 @@
+/**
+ * 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.MaybeObserver;
+import io.reactivex.MaybeSource;
+import io.reactivex.SingleObserver;
+import io.reactivex.SingleSource;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
+
+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 MaybeSwitchIfEmptyToSingle<T> extends AbstractMaybeWithUpstream<T, T> {
+
+    final SingleSource<? extends T> other;
+
+    public MaybeSwitchIfEmptyToSingle(MaybeSource<T> source, SingleSource<? extends T> other) {
+        super(source);
+        this.other = other;
+    }
+
+    @Override
+    protected void subscribeActual(MaybeObserver<? 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 MaybeObserver<? super T> actual;
+
+        final SingleSource<? extends T> other;
+
+        SwitchIfEmptyMaybeObserver(MaybeObserver<? 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 MaybeObserver<? super T> actual;
+
+            final AtomicReference<Disposable> parent;
+            OtherSingleObserver(MaybeObserver<? 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/MaybeSwitchIfEmptyTest.java → src/test/java/io/reactivex/internal/operators/maybe/MaybeSwitchIfEmptyToMaybeTest.java

@@ -24,7 +24,7 @@
 import io.reactivex.processors.PublishProcessor;
 import io.reactivex.schedulers.Schedulers;
 
-public class MaybeSwitchIfEmptyTest {
+public class MaybeSwitchIfEmptyToMaybeTest {
 
     @Test
     public void nonEmpty() {
@@ -36,16 +36,6 @@ public void empty() {
         Maybe.<Integer>empty().switchIfEmpty(Maybe.just(2)).test().assertResult(2);
     }
 
-    @Test
-    public void defaultIfEmptyNonEmpty() {
-        Maybe.just(1).defaultIfEmpty(2).test().assertResult(1);
-    }
-
-    @Test
-    public void defaultIfEmptyEmpty() {
-        Maybe.<Integer>empty().defaultIfEmpty(2).test().assertResult(2);
-    }
-
     @Test
     public void error() {
         Maybe.<Integer>error(new TestException()).switchIfEmpty(Maybe.just(2))

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

@@ -0,0 +1,118 @@
+/**
+ * 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 MaybeSwitchIfEmptyToSingleTest {
+
+    @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 defaultIfEmptyNonEmpty() {
+        Maybe.just(1).defaultIfEmpty(2).test().assertResult(1);
+    }
+
+    @Test
+    public void defaultIfEmptyEmpty() {
+        Maybe.<Integer>empty().defaultIfEmpty(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.checkDoubleOnSubscribeMaybe(new Function<Maybe<Integer>, Maybe<Integer>>() {
+            @Override
+            public Maybe<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());
+        }
+    }
+}