import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
private void handleScalarSynchronousObservable(ScalarSynchronousObservable<? extends T> t) {
    // fast-path for scalar, synchronous values such as Observable.from(int)
    /**
     * Without this optimization:
     * 
     * <pre> {@code
     * Benchmark                                          (size)   Mode   Samples        Score  Score error    Units
     * r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1         1  thrpt         5  2,418,452.409   130572.665    ops/s
     * r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1      1000  thrpt         5     5,690.456       94.958    ops/s
     * r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5          takes too long
     * 
     * With this optimization:
     * 
     * r.o.OperatorMergePerf.merge1SyncStreamOfN               1  thrpt         5  5,475,300.198   156741.334    ops/s
     * r.o.OperatorMergePerf.merge1SyncStreamOfN            1000  thrpt         5    68,932.278     1311.023    ops/s
     * r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5       64.405        0.611    ops/s
     * } </pre>
     * 
     */
    if (mergeProducer.requested == Long.MAX_VALUE) {
        handleScalarSynchronousObservableWithoutRequestLimits(t);
    } else {
        handleScalarSynchronousObservableWithRequestLimits(t);
    }
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
private void handleScalarSynchronousObservableWithoutRequestLimits(ScalarSynchronousObservable<? extends T> t) {
    T value = t.get();
    if (getEmitLock()) {
        try {
            actual.onNext(value);
            return;
        } finally {
            if (releaseEmitLock()) {
                drainQueuesIfNeeded();
            }
            request(1);
        }
    } else {
        initScalarValueQueueIfNeeded();
        try {
            scalarValueQueue.onNext(value);
        } catch (MissingBackpressureException e) {
            onError(e);
        }
        return;
    }
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
@Experimental
public final <R> Observable<R> concatMapDelayError(Func1<? super T, ? extends Observable<? extends R>> func) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarFlatMap(func);
    }
    return create(new OnSubscribeConcatMap(this, func, 2, 2));
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
@Beta
public final <R> Observable<R> flatMap(Func1<? super T, ? extends Observable<? extends R>> func, int maxConcurrent) {
    if (getClass() == ScalarSynchronousObservable.class) {
        return ((ScalarSynchronousObservable) this).scalarFlatMap(func);
    }
    return merge(map(func), maxConcurrent);
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public void onNext(Observable<? extends T> t) {
    if (t != null) {
        if (t instanceof ScalarSynchronousObservable) {
            tryEmit(((ScalarSynchronousObservable) t).get());
            return;
        }
        long j = this.uniqueId;
        this.uniqueId = 1 + j;
        InnerSubscriber<T> inner = new InnerSubscriber(this, j);
        addInner(inner);
        t.unsafeSubscribe(inner);
        emit();
    }
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
@Test
public void testIsNotScalarSynchronousObservableAfterFlatMap() {
    Observable<Integer> o = Observable.just(1).flatMap(new Func1<Integer, Observable<Integer>>() {

        @Override
        public Observable<Integer> call(Integer n) {
            return Observable.range(1, 100);
        }});
    Assert.assertFalse(o instanceof ScalarSynchronousObservable);
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
@Override
public void onNext(Observable<? extends T> t) {
    if (t instanceof ScalarSynchronousObservable) {
        handleScalarSynchronousObservable((ScalarSynchronousObservable)t);
    } else {
        if (t == null || isUnsubscribed()) {
            return;
        }
        synchronized (this) {
            // synchronized here because `wip` can be concurrently changed by children Observables
            wip++;
        }
        handleNewSource(t);
    }
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
private void handleScalarSynchronousObservableWithRequestLimits(ScalarSynchronousObservable<? extends T> t) {
    if (getEmitLock()) {
        boolean emitted = false;
        try {
            long r = mergeProducer.requested;
            if (r > 0) {
                emitted = true;
                actual.onNext(t.get());
                MergeProducer.REQUESTED.decrementAndGet(mergeProducer);
                // we handle this Observable without ever incrementing the wip or touching other machinery so just return here
                return;
            }
        } finally {
            if (releaseEmitLock()) {
                drainQueuesIfNeeded();
            }
            if (emitted) {
                request(1);
            }
        }
    }

    // if we didn't return above we need to enqueue
    // enqueue the values for later delivery
    initScalarValueQueueIfNeeded();
    try {
        scalarValueQueue.onNext(t.get());
    } catch (MissingBackpressureException e) {
        onError(e);
    }
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public static <T> Observable<T> just(T value) {
    return ScalarSynchronousObservable.create(value);
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public static <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source) {
    if (source.getClass() == ScalarSynchronousObservable.class) {
        return ((ScalarSynchronousObservable) source).scalarFlatMap(UtilityFunctions.identity());
    }
    return source.lift(OperatorMerge.instance(false));
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public final <R> Observable<R> concatMap(Func1<? super T, ? extends Observable<? extends R>> func) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarFlatMap(func);
    }
    return create(new OnSubscribeConcatMap(this, func, 2, 0));
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarScheduleOn(scheduler);
    }
    return lift(new OperatorObserveOn(scheduler, delayError, bufferSize));
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public final Observable<T> subscribeOn(Scheduler scheduler) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarScheduleOn(scheduler);
    }
    return create(new OperatorSubscribeOn(this, scheduler));
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public static final <T> Observable<T> just(T value) {
    return ScalarSynchronousObservable.create(value);
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public static final <T> Observable<T> merge(Observable<? extends Observable<? extends T>> source) {
    if (source.getClass() == ScalarSynchronousObservable.class) {
        return ((ScalarSynchronousObservable) source).scalarFlatMap(UtilityFunctions.identity());
    }
    return source.lift(OperatorMerge.instance(false));
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public final Observable<T> observeOn(Scheduler scheduler) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarScheduleOn(scheduler);
    }
    return lift(new OperatorObserveOn(scheduler));
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
public final Observable<T> subscribeOn(Scheduler scheduler) {
    if (this instanceof ScalarSynchronousObservable) {
        return ((ScalarSynchronousObservable) this).scalarScheduleOn(scheduler);
    }
    return nest().lift(new OperatorSubscribeOn(scheduler));
}
 

import rx.internal.util.ScalarSynchronousObservable; //导入依赖的package包/类
/**
 * Returns an Observable that emits a single item and then completes.
 * <p>
 * <img width="640" height="310" src="https://raw.github.com/wiki/Netflix/RxJava/images/rx-operators/just.png" alt="">
 * <p>
 * To convert any object into an Observable that emits that object, pass that object into the {@code just}
 * method.
 * <p>
 * This is similar to the {@link #from(java.lang.Object[])} method, except that {@code from} will convert
 * an {@link Iterable} object into an Observable that emits each of the items in the Iterable, one at a
 * time, while the {@code just} method converts an Iterable into an Observable that emits the entire
 * Iterable as a single item.
 * <dl>
 *  <dt><b>Scheduler:</b></dt>
 *  <dd>{@code just} does not operate by default on a particular {@link Scheduler}.</dd>
 * </dl>
 * 
 * @param value
 *            the item to emit
 * @param <T>
 *            the type of that item
 * @return an Observable that emits {@code value} as a single item and then completes
 * @see <a href="https://github.com/Netflix/RxJava/wiki/Creating-Observables#just">RxJava wiki: just</a>
 */
public final static <T> Observable<T> just(final T value) {
    return ScalarSynchronousObservable.create(value);
}