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.*;
@@ -654,15 +656,7 @@ public void call(SingleSubscriber<? super T> singleSubscriber) {
      * @see <a href="http://reactivex.io/documentation/operators/just.html">ReactiveX operators documentation: Just</a>
      */
     public 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);
     }
 
     /**
@@ -683,6 +677,9 @@ public void call(SingleSubscriber<? super T> te) {
      * @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>
      */
     public static <T> Single<T> merge(final Single<? extends Single<? extends T>> source) {
+        if (source instanceof ScalarSynchronousSingle) {
+            return ((ScalarSynchronousSingle<T>) source).scalarFlatMap((Func1) UtilityFunctions.identity());
+        }
         return Single.create(new OnSubscribe<T>() {
 
             @Override
@@ -1296,6 +1293,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 (this instanceof ScalarSynchronousSingle) {
+            return ((ScalarSynchronousSingle<T>) this).scalarFlatMap(func);
+        }
         return merge(map(func));
     }
 
@@ -1378,6 +1378,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));
     }
 
@@ -1737,6 +1740,9 @@ public void onNext(T t) {
      * @see #observeOn
      */
     public final Single<T> subscribeOn(final Scheduler scheduler) {
+        if (this instanceof ScalarSynchronousSingle) {
+            return ((ScalarSynchronousSingle<T>)this).scalarScheduleOn(scheduler);
+        }
         return create(new OnSubscribe<T>() {
             @Override
             public void call(final SingleSubscriber<? super T> t) {
@@ -1772,7 +1778,7 @@ public void onError(Throwable error) {
                     }
                 });
             }
-        });  
+        });
     }
 
     /**

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

@@ -0,0 +1,157 @@
+/**
+ * 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 value;
+
+    protected ScalarSynchronousSingle(final T t) {
+        super(new OnSubscribe<T>() {
+
+            @Override
+            public void call(SingleSubscriber<? super T> te) {
+                te.onSuccess(t);
+            }
+
+        });
+        this.value = t;
+    }
+
+    public T get() {
+        return value;
+    }
+
+    /**
+     * 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, value));
+        }
+        return create(new NormalScheduledEmission<T>(scheduler, value));
+    }
+
+    /**
+     * 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);
+            }
+        }
+    }
+
+    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(value);
+                if (o instanceof ScalarSynchronousSingle) {
+                    child.onSuccess(((ScalarSynchronousSingle<? extends R>) o).value);
+                } else {
+                    Subscriber<R> subscriber = 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);
+                        }
+                    };
+                    child.add(subscriber);
+                    o.unsafeSubscribe(subscriber);
+                }
+            }
+        });
+    }
+}

src/test/java/rx/SingleTest.java

@@ -1,11 +1,11 @@
 /**
  * Copyright 2015 Netflix, Inc.
- * 
+ *
  * 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.