2.x: Improve the scalar source performance of Observable.(concat|switch)MapX

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

@@ -14,6 +14,7 @@
 package io.reactivex.internal.operators.maybe;
 
 import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.fuseable.HasUpstreamMaybeSource;
@@ -40,17 +41,29 @@ public MaybeSource<T> source() {
 
     @Override
     protected void subscribeActual(Observer<? super T> s) {
-        source.subscribe(new MaybeToFlowableSubscriber<T>(s));
+        source.subscribe(create(s));
     }
 
-    static final class MaybeToFlowableSubscriber<T> extends DeferredScalarDisposable<T>
+    /**
+     * Creates a {@link MaybeObserver} wrapper around a {@link Observer}.
+     * @param <T> the value type
+     * @param downstream the downstream {@code Observer} to talk to
+     * @return the new MaybeObserver instance
+     * @since 2.1.11 - experimental
+     */
+    @Experimental
+    public static <T> MaybeObserver<T> create(Observer<? super T> downstream) {
+        return new MaybeToObservableObserver<T>(downstream);
+    }
+
+    static final class MaybeToObservableObserver<T> extends DeferredScalarDisposable<T>
     implements MaybeObserver<T> {
 
         private static final long serialVersionUID = 7603343402964826922L;
 
         Disposable d;
 
-        MaybeToFlowableSubscriber(Observer<? super T> actual) {
+        MaybeToObservableObserver(Observer<? super T> actual) {
             super(actual);
         }
 

src/main/java/io/reactivex/internal/operators/mixed/ObservableConcatMapCompletable.java

@@ -13,15 +13,14 @@
 
 package io.reactivex.internal.operators.mixed;
 
-import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.*;
 
 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.*;
+import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.functions.ObjectHelper;
 import io.reactivex.internal.fuseable.*;
 import io.reactivex.internal.queue.SpscLinkedArrayQueue;
@@ -57,7 +56,7 @@ public ObservableConcatMapCompletable(Observable<T> source,
 
     @Override
     protected void subscribeActual(CompletableObserver s) {
-        if (!tryScalarSource(source, mapper, s)) {
+        if (!ScalarXMapZHelper.tryAsCompletable(source, mapper, s)) {
             source.subscribe(new ConcatMapCompletableObserver<T>(s, mapper, errorMode, prefetch));
         }
     }
@@ -301,30 +300,4 @@ void dispose() {
             }
         }
     }
-
-    static <T> boolean tryScalarSource(Observable<T> source, Function<? super T, ? extends CompletableSource> mapper, CompletableObserver observer) {
-        if (source instanceof Callable) {
-            @SuppressWarnings("unchecked")
-            Callable<T> call = (Callable<T>) source;
-            CompletableSource cs = null;
-            try {
-                T item = call.call();
-                if (item != null) {
-                    cs = ObjectHelper.requireNonNull(mapper.apply(item), "The mapper returned a null CompletableSource");
-                }
-            } catch (Throwable ex) {
-                Exceptions.throwIfFatal(ex);
-                EmptyDisposable.error(ex, observer);
-                return true;
-            }
-
-            if (cs == null) {
-                EmptyDisposable.complete(observer);
-            } else {
-                cs.subscribe(observer);
-            }
-            return true;
-        }
-        return false;
-    }
 }

src/main/java/io/reactivex/internal/operators/mixed/ObservableConcatMapMaybe.java

@@ -59,7 +59,9 @@ public ObservableConcatMapMaybe(Observable<T> source,
 
     @Override
     protected void subscribeActual(Observer<? super R> s) {
-        source.subscribe(new ConcatMapMaybeMainObserver<T, R>(s, mapper, prefetch, errorMode));
+        if (!ScalarXMapZHelper.tryAsMaybe(source, mapper, s)) {
+            source.subscribe(new ConcatMapMaybeMainObserver<T, R>(s, mapper, prefetch, errorMode));
+        }
     }
 
     static final class ConcatMapMaybeMainObserver<T, R>

src/main/java/io/reactivex/internal/operators/mixed/ObservableConcatMapSingle.java

@@ -59,7 +59,9 @@ public ObservableConcatMapSingle(Observable<T> source,
 
     @Override
     protected void subscribeActual(Observer<? super R> s) {
-        source.subscribe(new ConcatMapSingleMainObserver<T, R>(s, mapper, prefetch, errorMode));
+        if (!ScalarXMapZHelper.tryAsSingle(source, mapper, s)) {
+            source.subscribe(new ConcatMapSingleMainObserver<T, R>(s, mapper, prefetch, errorMode));
+        }
     }
 
     static final class ConcatMapSingleMainObserver<T, R>

src/main/java/io/reactivex/internal/operators/mixed/ObservableSwitchMapCompletable.java

@@ -51,7 +51,9 @@ public ObservableSwitchMapCompletable(Observable<T> source,
 
     @Override
     protected void subscribeActual(CompletableObserver s) {
-        source.subscribe(new SwitchMapCompletableObserver<T>(s, mapper, delayErrors));
+        if (!ScalarXMapZHelper.tryAsCompletable(source, mapper, s)) {
+            source.subscribe(new SwitchMapCompletableObserver<T>(s, mapper, delayErrors));
+        }
     }
 
     static final class SwitchMapCompletableObserver<T> implements Observer<T>, Disposable {

src/main/java/io/reactivex/internal/operators/mixed/ObservableSwitchMapMaybe.java

@@ -53,7 +53,9 @@ public ObservableSwitchMapMaybe(Observable<T> source,
 
     @Override
     protected void subscribeActual(Observer<? super R> s) {
-        source.subscribe(new SwitchMapMaybeMainObserver<T, R>(s, mapper, delayErrors));
+        if (!ScalarXMapZHelper.tryAsMaybe(source, mapper, s)) {
+            source.subscribe(new SwitchMapMaybeMainObserver<T, R>(s, mapper, delayErrors));
+        }
     }
 
     static final class SwitchMapMaybeMainObserver<T, R> extends AtomicInteger

src/main/java/io/reactivex/internal/operators/mixed/ObservableSwitchMapSingle.java

@@ -53,7 +53,9 @@ public ObservableSwitchMapSingle(Observable<T> source,
 
     @Override
     protected void subscribeActual(Observer<? super R> s) {
-        source.subscribe(new SwitchMapSingleMainObserver<T, R>(s, mapper, delayErrors));
+        if (!ScalarXMapZHelper.tryAsSingle(source, mapper, s)) {
+            source.subscribe(new SwitchMapSingleMainObserver<T, R>(s, mapper, delayErrors));
+        }
     }
 
     static final class SwitchMapSingleMainObserver<T, R> extends AtomicInteger

src/main/java/io/reactivex/internal/operators/mixed/ScalarXMapZHelper.java

@@ -0,0 +1,156 @@
+/**
+ * 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.mixed;
+
+import java.util.concurrent.Callable;
+
+import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
+import io.reactivex.exceptions.Exceptions;
+import io.reactivex.functions.Function;
+import io.reactivex.internal.disposables.EmptyDisposable;
+import io.reactivex.internal.functions.ObjectHelper;
+import io.reactivex.internal.operators.maybe.MaybeToObservable;
+import io.reactivex.internal.operators.single.SingleToObservable;
+
+/**
+ * Utility class to extract a value from a scalar source reactive type,
+ * map it to a 0-1 type then subscribe the output type's consumer to it,
+ * saving on the overhead of the regular subscription channel.
+ * @since 2.1.11 - experimental
+ */
+@Experimental
+final class ScalarXMapZHelper {
+
+    private ScalarXMapZHelper() {
+        throw new IllegalStateException("No instances!");
+    }
+
+    /**
+     * Try subscribing to a {@link CompletableSource} mapped from
+     * a scalar source (which implements {@link Callable}).
+     * @param <T> the upstream value type
+     * @param source the source reactive type ({@code Flowable} or {@code Observable})
+     *               possibly implementing {@link Callable}.
+     * @param mapper the function that turns the scalar upstream value into a
+     *              {@link CompletableSource}
+     * @param observer the consumer to subscribe to the mapped {@link CompletableSource}
+     * @return true if a subscription did happen and the regular path should be skipped
+     */
+    static <T> boolean tryAsCompletable(Object source,
+            Function<? super T, ? extends CompletableSource> mapper,
+            CompletableObserver observer) {
+        if (source instanceof Callable) {
+            @SuppressWarnings("unchecked")
+            Callable<T> call = (Callable<T>) source;
+            CompletableSource cs = null;
+            try {
+                T item = call.call();
+                if (item != null) {
+                    cs = ObjectHelper.requireNonNull(mapper.apply(item), "The mapper returned a null CompletableSource");
+                }
+            } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
+                EmptyDisposable.error(ex, observer);
+                return true;
+            }
+
+            if (cs == null) {
+                EmptyDisposable.complete(observer);
+            } else {
+                cs.subscribe(observer);
+            }
+            return true;
+        }
+        return false;
+    }
+
+    /**
+     * Try subscribing to a {@link MaybeSource} mapped from
+     * a scalar source (which implements {@link Callable}).
+     * @param <T> the upstream value type
+     * @param source the source reactive type ({@code Flowable} or {@code Observable})
+     *               possibly implementing {@link Callable}.
+     * @param mapper the function that turns the scalar upstream value into a
+     *              {@link MaybeSource}
+     * @param observer the consumer to subscribe to the mapped {@link MaybeSource}
+     * @return true if a subscription did happen and the regular path should be skipped
+     */
+    static <T, R> boolean tryAsMaybe(Object source,
+            Function<? super T, ? extends MaybeSource<? extends R>> mapper,
+            Observer<? super R> observer) {
+        if (source instanceof Callable) {
+            @SuppressWarnings("unchecked")
+            Callable<T> call = (Callable<T>) source;
+            MaybeSource<? extends R> cs = null;
+            try {
+                T item = call.call();
+                if (item != null) {
+                    cs = ObjectHelper.requireNonNull(mapper.apply(item), "The mapper returned a null MaybeSource");
+                }
+            } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
+                EmptyDisposable.error(ex, observer);
+                return true;
+            }
+
+            if (cs == null) {
+                EmptyDisposable.complete(observer);
+            } else {
+                cs.subscribe(MaybeToObservable.create(observer));
+            }
+            return true;
+        }
+        return false;
+    }
+
+    /**
+     * Try subscribing to a {@link SingleSource} mapped from
+     * a scalar source (which implements {@link Callable}).
+     * @param <T> the upstream value type
+     * @param source the source reactive type ({@code Flowable} or {@code Observable})
+     *               possibly implementing {@link Callable}.
+     * @param mapper the function that turns the scalar upstream value into a
+     *              {@link SingleSource}
+     * @param observer the consumer to subscribe to the mapped {@link SingleSource}
+     * @return true if a subscription did happen and the regular path should be skipped
+     */
+    static <T, R> boolean tryAsSingle(Object source,
+            Function<? super T, ? extends SingleSource<? extends R>> mapper,
+            Observer<? super R> observer) {
+        if (source instanceof Callable) {
+            @SuppressWarnings("unchecked")
+            Callable<T> call = (Callable<T>) source;
+            SingleSource<? extends R> cs = null;
+            try {
+                T item = call.call();
+                if (item != null) {
+                    cs = ObjectHelper.requireNonNull(mapper.apply(item), "The mapper returned a null SingleSource");
+                }
+            } catch (Throwable ex) {
+                Exceptions.throwIfFatal(ex);
+                EmptyDisposable.error(ex, observer);
+                return true;
+            }
+
+            if (cs == null) {
+                EmptyDisposable.complete(observer);
+            } else {
+                cs.subscribe(SingleToObservable.create(observer));
+            }
+            return true;
+        }
+        return false;
+    }
+}

src/main/java/io/reactivex/internal/operators/single/SingleToObservable.java

@@ -13,6 +13,7 @@
 package io.reactivex.internal.operators.single;
 
 import io.reactivex.*;
+import io.reactivex.annotations.Experimental;
 import io.reactivex.disposables.Disposable;
 import io.reactivex.internal.disposables.DisposableHelper;
 
@@ -31,7 +32,19 @@ public SingleToObservable(SingleSource<? extends T> source) {
 
     @Override
     public void subscribeActual(final Observer<? super T> s) {
-        source.subscribe(new SingleToObservableObserver<T>(s));
+        source.subscribe(create(s));
+    }
+
+    /**
+     * Creates a {@link SingleObserver} wrapper around a {@link Observer}.
+     * @param <T> the value type
+     * @param downstream the downstream {@code Observer} to talk to
+     * @return the new SingleObserver instance
+     * @since 2.1.11 - experimental
+     */
+    @Experimental
+    public static <T> SingleObserver<T> create(Observer<? super T> downstream) {
+        return new SingleToObservableObserver<T>(downstream);
     }
 
     static final class SingleToObservableObserver<T>

src/test/java/io/reactivex/internal/operators/mixed/ObservableConcatMapMaybeTest.java

@@ -340,10 +340,37 @@ public MaybeSource<? extends Object> apply(Integer v)
         assertFalse(ps.hasObservers());
     }
 
+    @Test
+    public void scalarMapperCrash() {
+        TestObserver<Object> to = Observable.just(1)
+        .concatMapMaybe(new Function<Integer, MaybeSource<? extends Object>>() {
+            @Override
+            public MaybeSource<? extends Object> apply(Integer v)
+                    throws Exception {
+                        throw new TestException();
+                    }
+        })
+        .test();
+
+        to.assertFailure(TestException.class);
+    }
+
     @Test
     public void disposed() {
-        TestHelper.checkDisposed(Observable.just(1)
+        TestHelper.checkDisposed(Observable.just(1).hide()
                 .concatMapMaybe(Functions.justFunction(Maybe.never()))
         );
     }
+
+    @Test
+    public void scalarEmptySource() {
+        MaybeSubject<Integer> ms = MaybeSubject.create();
+
+        Observable.empty()
+        .concatMapMaybe(Functions.justFunction(ms))
+        .test()
+        .assertResult();
+
+        assertFalse(ms.hasObservers());
+    }
 }