1.x: Optimizate single just

src/main/java/rx/Single.java

@@ -37,6 +37,8 @@
 import rx.annotations.Beta;
 import rx.internal.operators.*;
 import rx.internal.producers.SingleDelayedProducer;
+import rx.internal.util.ScalarSynchronousSingle;
+import rx.internal.util.UtilityFunctions;
 import rx.singles.BlockingSingle;
 import rx.observers.SafeSubscriber;
 import rx.plugins.*;
@@ -656,15 +658,7 @@ public void call(SingleSubscriber<? super T> singleSubscriber) {
      * @see <a href="http://reactivex.io/documentation/operators/just.html">ReactiveX operators documentation: Just</a>
      */
     public final static <T> Single<T> just(final T value) {
-        // TODO add similar optimization as ScalarSynchronousObservable
-        return Single.create(new OnSubscribe<T>() {
-
-            @Override
-            public void call(SingleSubscriber<? super T> te) {
-                te.onSuccess(value);
-            }
-
-        });
+        return ScalarSynchronousSingle.create(value);
     }
 
     /**
@@ -685,6 +679,9 @@ public void call(SingleSubscriber<? super T> te) {
      * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>
      */
     public final static <T> Single<T> merge(final Single<? extends Single<? extends T>> source) {
+        if (source.getClass() == ScalarSynchronousSingle.class) {
+            return ((ScalarSynchronousSingle<T>)source).scalarFlatMap((Func1) UtilityFunctions.identity());
+        }
         return Single.create(new OnSubscribe<T>() {
 
             @Override
@@ -1258,6 +1255,9 @@ public final Observable<T> concatWith(Single<? extends T> t1) {
      * @see <a href="http://reactivex.io/documentation/operators/flatmap.html">ReactiveX operators documentation: FlatMap</a>
      */
     public final <R> Single<R> flatMap(final Func1<? super T, ? extends Single<? extends R>> func) {
+        if (getClass() == ScalarSynchronousSingle.class) {
+            return ((ScalarSynchronousSingle<T>)this).scalarFlatMap(func);
+        }
         return merge(map(func));
     }
 
@@ -1340,6 +1340,9 @@ public final Observable<T> mergeWith(Single<? extends T> t1) {
      * @see #subscribeOn
      */
     public final Single<T> observeOn(Scheduler scheduler) {
+        if (this instanceof ScalarSynchronousSingle) {
+            return ((ScalarSynchronousSingle<T>)this).scalarScheduleOn(scheduler);
+        }
         return lift(new OperatorObserveOn<T>(scheduler));
     }
 
@@ -1699,6 +1702,9 @@ public void onNext(T t) {
      * @see #observeOn
      */
     public final Single<T> subscribeOn(Scheduler scheduler) {
+        if (this instanceof ScalarSynchronousSingle) {
+            return ((ScalarSynchronousSingle<T>)this).scalarScheduleOn(scheduler);
+        }
         return nest().lift(new OperatorSubscribeOn<T>(scheduler));
     }
 

src/main/java/rx/internal/util/ScalarSynchronousSingle.java

@@ -0,0 +1,156 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * <p/>
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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 rx.internal.util;
+
+import rx.Scheduler;
+import rx.Scheduler.Worker;
+import rx.Single;
+import rx.SingleSubscriber;
+import rx.Subscriber;
+import rx.functions.Action0;
+import rx.functions.Func1;
+import rx.internal.schedulers.EventLoopsScheduler;
+
+public final class ScalarSynchronousSingle<T> extends Single<T> {
+
+    public static final <T> ScalarSynchronousSingle<T> create(T t) {
+        return new ScalarSynchronousSingle<T>(t);
+    }
+
+    final T t;
+
+    protected ScalarSynchronousSingle(final T t) {
+        super(new OnSubscribe<T>() {
+
+            @Override
+            public void call(SingleSubscriber<? super T> te) {
+                te.onSuccess(t);
+            }
+
+        });
+        this.t = t;
+    }
+
+    public T get() {
+        return t;
+    }
+
+    /**
+     * Customized observeOn/subscribeOn implementation which emits the scalar
+     * value directly or with less overhead on the specified scheduler.
+     *
+     * @param scheduler the target scheduler
+     * @return the new observable
+     */
+    public Single<T> scalarScheduleOn(Scheduler scheduler) {
+        if (scheduler instanceof EventLoopsScheduler) {
+            EventLoopsScheduler es = (EventLoopsScheduler) scheduler;
+            return create(new DirectScheduledEmission<T>(es, t));
+        }
+        return create(new NormalScheduledEmission<T>(scheduler, t));
+    }
+
+    /**
+     * Optimized observeOn for scalar value observed on the EventLoopsScheduler.
+     */
+    static final class DirectScheduledEmission<T> implements OnSubscribe<T> {
+        private final EventLoopsScheduler es;
+        private final T value;
+
+        DirectScheduledEmission(EventLoopsScheduler es, T value) {
+            this.es = es;
+            this.value = value;
+        }
+
+        @Override
+        public void call(SingleSubscriber<? super T> singleSubscriber) {
+            singleSubscriber.add(es.scheduleDirect(new ScalarSynchronousSingleAction<T>(singleSubscriber, value)));
+        }
+    }
+
+    /**
+     * Emits a scalar value on a general scheduler.
+     */
+    static final class NormalScheduledEmission<T> implements OnSubscribe<T> {
+        private final Scheduler scheduler;
+        private final T value;
+
+        NormalScheduledEmission(Scheduler scheduler, T value) {
+            this.scheduler = scheduler;
+            this.value = value;
+        }
+
+        @Override
+        public void call(SingleSubscriber<? super T> singleSubscriber) {
+            Worker worker = scheduler.createWorker();
+            singleSubscriber.add(worker);
+            worker.schedule(new ScalarSynchronousSingleAction<T>(singleSubscriber, value));
+        }
+    }
+
+    /**
+     * Action that emits a single value when called.
+     */
+    static final class ScalarSynchronousSingleAction<T> implements Action0 {
+        private final SingleSubscriber<? super T> subscriber;
+        private final T value;
+
+        ScalarSynchronousSingleAction(SingleSubscriber<? super T> subscriber,
+                                      T value) {
+            this.subscriber = subscriber;
+            this.value = value;
+        }
+
+        @Override
+        public void call() {
+            try {
+                subscriber.onSuccess(value);
+            } catch (Throwable t) {
+                subscriber.onError(t);
+                return;
+            }
+        }
+    }
+
+    public <R> Single<R> scalarFlatMap(final Func1<? super T, ? extends Single<? extends R>> func) {
+        return create(new OnSubscribe<R>() {
+            @Override
+            public void call(final SingleSubscriber<? super R> child) {
+
+                Single<? extends R> o = func.call(t);
+                if (o.getClass() == ScalarSynchronousSingle.class) {
+                    child.onSuccess(((ScalarSynchronousSingle<? extends R>) o).t);
+                } else {
+                    o.unsafeSubscribe(new Subscriber<R>() {
+                        @Override
+                        public void onCompleted() {
+                        }
+
+                        @Override
+                        public void onError(Throwable e) {
+                            child.onError(e);
+                        }
+
+                        @Override
+                        public void onNext(R r) {
+                            child.onSuccess(r);
+                        }
+                    });
+                }
+            }
+        });
+    }
+}