/**

* 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;

import java.util.NoSuchElementException;

import java.util.concurrent.*;

import org.reactivestreams.Publisher;

import io.reactivex.annotations.*;

import io.reactivex.disposables.Disposable;

import io.reactivex.exceptions.Exceptions;

import io.reactivex.functions.*;

import io.reactivex.internal.functions.*;

import io.reactivex.internal.fuseable.*;

import io.reactivex.internal.observers.*;

import io.reactivex.internal.operators.completable.*;

import io.reactivex.internal.operators.flowable.*;

import io.reactivex.internal.operators.maybe.*;

import io.reactivex.internal.operators.observable.*;

import io.reactivex.internal.operators.single.*;

import io.reactivex.internal.util.*;

import io.reactivex.observers.TestObserver;

import io.reactivex.plugins.RxJavaPlugins;

import io.reactivex.schedulers.Schedulers;

/**

* The Single class implements the Reactive Pattern for a single value response.

* See {@link Flowable} or {@link Observable} for the

* implementation of the Reactive Pattern for a stream or vector of values.

* <p>

* {@code Single} behaves the same as {@link Observable} except that it can only emit either a single successful

* value, or an error (there is no "onComplete" notification as there is for {@link Observable})

* <p>

* Like an {@link Observable}, a {@code Single} is lazy, can be either "hot" or "cold", synchronous or

* asynchronous.

* <p>

* The documentation for this class makes use of marble diagrams. The following legend explains these diagrams:

* <p>

* <img width="640" height="301" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.legend.png" alt="">

* <p>

* For more information see the <a href="http://reactivex.io/documentation/observable.html">ReactiveX

* documentation</a>.

*

* @param <T>

* the type of the item emitted by the Single

* @since 2.0

*/

public abstract class Single<T> implements SingleSource<T> {

/**

* Runs multiple Single sources and signals the events of the first one that signals (cancelling

* the rest).

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code amb} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param sources the Iterable sequence of sources. A subscription to each source will

* occur in the same order as in this Iterable.

* @return the new Single instance

* @since 2.0

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> amb(final Iterable<? extends SingleSource<? extends T>> sources) {

ObjectHelper.requireNonNull(sources, "sources is null");

return RxJavaPlugins.onAssembly(new SingleAmb<T>(null, sources));

}

/**

* Runs multiple Single sources and signals the events of the first one that signals (cancelling

* the rest).

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code ambArray} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param sources the array of sources. A subscription to each source will

* occur in the same order as in this array.

* @return the new Single instance

* @since 2.0

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Single<T> ambArray(final SingleSource<? extends T>... sources) {

if (sources.length == 0) {

return error(SingleInternalHelper.<T>emptyThrower());

}

if (sources.length == 1) {

return wrap((SingleSource<T>)sources[0]);

}

return RxJavaPlugins.onAssembly(new SingleAmb<T>(sources, null));

}

/**

* Concatenate the single values, in a non-overlapping fashion, of the Single sources provided by

* an Iterable sequence.

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code concat} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param sources the Iterable sequence of SingleSource instances

* @return the new Flowable instance

* @since 2.0

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@BackpressureSupport(BackpressureKind.FULL)

public static <T> Flowable<T> concat(Iterable<? extends SingleSource<? extends T>> sources) {

return concat(Flowable.fromIterable(sources));

}

/**

* Concatenate the single values, in a non-overlapping fashion, of the Single sources provided by

* an Observable sequence.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code concat} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param sources the ObservableSource of SingleSource instances

* @return the new Observable instance

* @since 2.0

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings({ "unchecked", "rawtypes" })

public static <T> Observable<T> concat(ObservableSource<? extends SingleSource<? extends T>> sources) {

ObjectHelper.requireNonNull(sources, "sources is null");

return RxJavaPlugins.onAssembly(new ObservableConcatMap(sources, SingleInternalHelper.toObservable(), 2, ErrorMode.IMMEDIATE));

}

/**

* Concatenate the single values, in a non-overlapping fashion, of the Single sources provided by

* a Publisher sequence.

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer

* and the sources {@code Publisher} is expected to honor it as well.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code concat} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param sources the Publisher of SingleSource instances

* @return the new Flowable instance

* @since 2.0

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Flowable<T> concat(Publisher<? extends SingleSource<? extends T>> sources) {

return concat(sources, 2);

}

/**

* Concatenate the single values, in a non-overlapping fashion, of the Single sources provided by

* a Publisher sequence and prefetched by the specified amount.

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer

* and the sources {@code Publisher} is expected to honor it as well.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code concat} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param sources the Publisher of SingleSource instances

* @param prefetch the number of SingleSources to prefetch from the Publisher

* @return the new Flowable instance

* @since 2.0

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings({ "unchecked", "rawtypes" })

public static <T> Flowable<T> concat(Publisher<? extends SingleSource<? extends T>> sources, int prefetch) {

ObjectHelper.requireNonNull(sources, "sources is null");

ObjectHelper.verifyPositive(prefetch, "prefetch");

return RxJavaPlugins.onAssembly(new FlowableConcatMapPublisher(sources, SingleInternalHelper.toFlowable(), prefetch, ErrorMode.IMMEDIATE));

}

/**

* Returns a Flowable that emits the items emitted by two Singles, one after the other.

* <p>

* <img width="640" height="380" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.concat.png" alt="">

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code concat} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param <T> the common value type

* @param source1

* a Single to be concatenated

* @param source2

* a Single to be concatenated

* @return a Flowable that emits items emitted by the two source Singles, one after the other.

* @see <a href="http://reactivex.io/documentation/operators/concat.html">ReactiveX operators documentation: Concat</a>

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> concat(

SingleSource<? extends T> source1, SingleSource<? extends T> source2

) {

ObjectHelper.requireNonNull(source1, "source1 is null");

ObjectHelper.requireNonNull(source2, "source2 is null");

return concat(Flowable.fromArray(source1, source2));

}

/**

* Returns a Flowable that emits the items emitted by three Singles, one after the other.

* <p>

* <img width="640" height="380" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.concat.png" alt="">

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code concat} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param <T> the common value type

* @param source1

* a Single to be concatenated

* @param source2

* a Single to be concatenated

* @param source3

* a Single to be concatenated

* @return a Flowable that emits items emitted by the three source Singles, one after the other.

* @see <a href="http://reactivex.io/documentation/operators/concat.html">ReactiveX operators documentation: Concat</a>

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> concat(

SingleSource<? extends T> source1, SingleSource<? extends T> source2,

SingleSource<? extends T> source3

) {

ObjectHelper.requireNonNull(source1, "source1 is null");

ObjectHelper.requireNonNull(source2, "source2 is null");

ObjectHelper.requireNonNull(source3, "source3 is null");

return concat(Flowable.fromArray(source1, source2, source3));

}

/**

* Returns a Flowable that emits the items emitted by four Singles, one after the other.

* <p>

* <img width="640" height="380" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.concat.png" alt="">

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code concat} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param <T> the common value type

* @param source1

* a Single to be concatenated

* @param source2

* a Single to be concatenated

* @param source3

* a Single to be concatenated

* @param source4

* a Single to be concatenated

* @return a Flowable that emits items emitted by the four source Singles, one after the other.

* @see <a href="http://reactivex.io/documentation/operators/concat.html">ReactiveX operators documentation: Concat</a>

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> concat(

SingleSource<? extends T> source1, SingleSource<? extends T> source2,

SingleSource<? extends T> source3, SingleSource<? extends T> source4

) {

ObjectHelper.requireNonNull(source1, "source1 is null");

ObjectHelper.requireNonNull(source2, "source2 is null");

ObjectHelper.requireNonNull(source3, "source3 is null");

ObjectHelper.requireNonNull(source4, "source4 is null");

return concat(Flowable.fromArray(source1, source2, source3, source4));

}

/**

* Concatenate the single values, in a non-overlapping fashion, of the Single sources provided in

* an array.

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code concatArray} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param sources the array of SingleSource instances

* @return the new Flowable instance

* @since 2.0

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings({ "unchecked", "rawtypes" })

public static <T> Flowable<T> concatArray(SingleSource<? extends T>... sources) {

return RxJavaPlugins.onAssembly(new FlowableConcatMap(Flowable.fromArray(sources), SingleInternalHelper.toFlowable(), 2, ErrorMode.BOUNDARY));

}

/**

* Provides an API (via a cold Completable) that bridges the reactive world with the callback-style world.

* <p>

* Example:

* <pre><code>

* Single.&lt;Event&gt;create(emitter -&gt; {

* Callback listener = new Callback() {

* &#64;Override

* public void onEvent(Event e) {

* emitter.onSuccess(e);

* }

*

* &#64;Override

* public void onFailure(Exception e) {

* emitter.onError(e);

* }

* };

*

* AutoCloseable c = api.someMethod(listener);

*

* emitter.setCancellable(c::close);

*

* });

* </code></pre>

* <p>

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code create} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param source the emitter that is called when a SingleObserver subscribes to the returned {@code Single}

* @return the new Single instance

* @see SingleOnSubscribe

* @see Cancellable

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> create(SingleOnSubscribe<T> source) {

ObjectHelper.requireNonNull(source, "source is null");

return RxJavaPlugins.onAssembly(new SingleCreate<T>(source));

}

/**

* Calls a Callable for each individual SingleObserver to return the actual Single source to

* be subscribed to.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code defer} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param singleSupplier the Callable that is called for each individual SingleObserver and

* returns a SingleSource instance to subscribe to

* @return the new Single instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> defer(final Callable<? extends SingleSource<? extends T>> singleSupplier) {

ObjectHelper.requireNonNull(singleSupplier, "singleSupplier is null");

return RxJavaPlugins.onAssembly(new SingleDefer<T>(singleSupplier));

}

/**

* Signals a Throwable returned by the callback function for each individual SingleObserver.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code error} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param errorSupplier the callable that is called for each individual SingleObserver and

* returns a Throwable instance to be emitted.

* @return the new Single instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> error(final Callable<? extends Throwable> errorSupplier) {

ObjectHelper.requireNonNull(errorSupplier, "errorSupplier is null");

return RxJavaPlugins.onAssembly(new SingleError<T>(errorSupplier));

}

/**

* Returns a Single that invokes a subscriber's {@link SingleObserver#onError onError} method when the

* subscriber subscribes to it.

* <p>

* <img width="640" height="190" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.error.png" alt="">

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code error} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param exception

* the particular Throwable to pass to {@link SingleObserver#onError onError}

* @param <T>

* the type of the item (ostensibly) emitted by the Single

* @return a Single that invokes the subscriber's {@link SingleObserver#onError onError} method when

* the subscriber subscribes to it

* @see <a href="http://reactivex.io/documentation/operators/empty-never-throw.html">ReactiveX operators documentation: Throw</a>

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> error(final Throwable exception) {

ObjectHelper.requireNonNull(exception, "error is null");

return error(Functions.justCallable(exception));

}

/**

* Returns a {@link Single} that invokes passed function and emits its result for each new SingleObserver that subscribes.

* <p>

* Allows you to defer execution of passed function until SingleObserver subscribes to the {@link Single}.

* It makes passed function "lazy".

* Result of the function invocation will be emitted by the {@link Single}.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code fromCallable} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param callable

* function which execution should be deferred, it will be invoked when SingleObserver will subscribe to the {@link Single}.

* @param <T>

* the type of the item emitted by the {@link Single}.

* @return a {@link Single} whose {@link SingleObserver}s' subscriptions trigger an invocation of the given function.

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> fromCallable(final Callable<? extends T> callable) {

ObjectHelper.requireNonNull(callable, "callable is null");

return RxJavaPlugins.onAssembly(new SingleFromCallable<T>(callable));

}

/**

* Converts a {@link Future} into a {@code Single}.

* <p>

* <img width="640" height="315" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.from.Future.png" alt="">

* <p>

* You can convert any object that supports the {@link Future} interface into a Single that emits the return

* value of the {@link Future#get} method of that object, by passing the object into the {@code from}

* method.

* <p>

* <em>Important note:</em> This Single is blocking; you cannot dispose it.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code fromFuture} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param future

* the source {@link Future}

* @param <T>

* the type of object that the {@link Future} returns, and also the type of item to be emitted by

* the resulting {@code Single}

* @return a {@code Single} that emits the item from the source {@link Future}

* @see <a href="http://reactivex.io/documentation/operators/from.html">ReactiveX operators documentation: From</a>

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> fromFuture(Future<? extends T> future) {

return toSingle(Flowable.<T>fromFuture(future));

}

/**

* Converts a {@link Future} into a {@code Single}, with a timeout on the Future.

* <p>

* <img width="640" height="315" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.from.Future.png" alt="">

* <p>

* You can convert any object that supports the {@link Future} interface into a {@code Single} that emits

* the return value of the {@link Future#get} method of that object, by passing the object into the

* {@code from} method.

* <p>

* <em>Important note:</em> This {@code Single} is blocking; you cannot dispose it.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code fromFuture} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param future

* the source {@link Future}

* @param timeout

* the maximum time to wait before calling {@code get}

* @param unit

* the {@link TimeUnit} of the {@code timeout} argument

* @param <T>

* the type of object that the {@link Future} returns, and also the type of item to be emitted by

* the resulting {@code Single}

* @return a {@code Single} that emits the item from the source {@link Future}

* @see <a href="http://reactivex.io/documentation/operators/from.html">ReactiveX operators documentation: From</a>

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> fromFuture(Future<? extends T> future, long timeout, TimeUnit unit) {

return toSingle(Flowable.<T>fromFuture(future, timeout, unit));

}

/**

* Converts a {@link Future} into a {@code Single}, with a timeout on the Future.

* <p>

* <img width="640" height="315" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.from.Future.png" alt="">

* <p>

* You can convert any object that supports the {@link Future} interface into a {@code Single} that emits

* the return value of the {@link Future#get} method of that object, by passing the object into the

* {@code from} method.

* <p>

* <em>Important note:</em> This {@code Single} is blocking; you cannot dispose it.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>You specify the {@link Scheduler} where the blocking wait will happen.</dd>

* </dl>

*

* @param future

* the source {@link Future}

* @param timeout

* the maximum time to wait before calling {@code get}

* @param unit

* the {@link TimeUnit} of the {@code timeout} argument

* @param scheduler

* the Scheduler to use for the blocking wait

* @param <T>

* the type of object that the {@link Future} returns, and also the type of item to be emitted by

* the resulting {@code Single}

* @return a {@code Single} that emits the item from the source {@link Future}

* @see <a href="http://reactivex.io/documentation/operators/from.html">ReactiveX operators documentation: From</a>

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.CUSTOM)

public static <T> Single<T> fromFuture(Future<? extends T> future, long timeout, TimeUnit unit, Scheduler scheduler) {

return toSingle(Flowable.<T>fromFuture(future, timeout, unit, scheduler));

}

/**

* Converts a {@link Future}, operating on a specified {@link Scheduler}, into a {@code Single}.

* <p>

* <img width="640" height="315" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.from.Future.s.png" alt="">

* <p>

* You can convert any object that supports the {@link Future} interface into a {@code Single} that emits

* the return value of the {@link Future#get} method of that object, by passing the object into the

* {@code from} method.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>You specify which {@link Scheduler} this operator will use</dd>

* </dl>

*

* @param future

* the source {@link Future}

* @param scheduler

* the {@link Scheduler} to wait for the Future on. Use a Scheduler such as

* {@link Schedulers#io()} that can block and wait on the Future

* @param <T>

* the type of object that the {@link Future} returns, and also the type of item to be emitted by

* the resulting {@code Single}

* @return a {@code Single} that emits the item from the source {@link Future}

* @see <a href="http://reactivex.io/documentation/operators/from.html">ReactiveX operators documentation: From</a>

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.CUSTOM)

public static <T> Single<T> fromFuture(Future<? extends T> future, Scheduler scheduler) {

return toSingle(Flowable.<T>fromFuture(future, scheduler));

}

/**

* Wraps a specific Publisher into a Single and signals its single element or error.

* <p>If the source Publisher is empty, a NoSuchElementException is signalled. If

* the source has more than one element, an IndexOutOfBoundsException is signalled.

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The {@code publisher} is consumed in an unbounded fashion but will be cancelled

* if it produced more than one item.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code fromPublisher} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param publisher the source Publisher instance, not null

* @return the new Single instance

*/

@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> fromPublisher(final Publisher<? extends T> publisher) {

ObjectHelper.requireNonNull(publisher, "publisher is null");

return RxJavaPlugins.onAssembly(new SingleFromPublisher<T>(publisher));

}

/**

* Wraps a specific ObservableSource into a Single and signals its single element or error.

* <p>If the ObservableSource is empty, a NoSuchElementException is signalled.

* If the source has more than one element, an IndexOutOfBoundsException is signalled.

* <p>

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code fromObservable} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param observableSource the source Observable, not null

* @param <T>

* the type of the item emitted by the {@link Single}.

* @return the new Single instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> fromObservable(ObservableSource<? extends T> observableSource) {

ObjectHelper.requireNonNull(observableSource, "observableSource is null");

return RxJavaPlugins.onAssembly(new ObservableSingleSingle<T>(observableSource, null));

}

/**

* Returns a {@code Single} that emits a specified item.

* <p>

* <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.just.png" alt="">

* <p>

* To convert any object into a {@code Single} that emits that object, pass that object into the

* {@code just} method.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code just} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param item

* the item to emit

* @param <T>

* the type of that item

* @return a {@code Single} that emits {@code item}

* @see <a href="http://reactivex.io/documentation/operators/just.html">ReactiveX operators documentation: Just</a>

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> just(final T item) {

ObjectHelper.requireNonNull(item, "value is null");

return RxJavaPlugins.onAssembly(new SingleJust<T>(item));

}

/**

* Merges an Iterable sequence of SingleSource instances into a single Flowable sequence,

* running all SingleSources at once.

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code merge} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the common and resulting value type

* @param sources the Iterable sequence of SingleSource sources

* @return the new Flowable instance

* @since 2.0

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Flowable<T> merge(Iterable<? extends SingleSource<? extends T>> sources) {

return merge(Flowable.fromIterable(sources));

}

/**

* Merges a Flowable sequence of SingleSource instances into a single Flowable sequence,

* running all SingleSources at once.

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code merge} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the common and resulting value type

* @param sources the Flowable sequence of SingleSource sources

* @return the new Flowable instance

* @since 2.0

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings({ "unchecked", "rawtypes" })

public static <T> Flowable<T> merge(Publisher<? extends SingleSource<? extends T>> sources) {

ObjectHelper.requireNonNull(sources, "sources is null");

return RxJavaPlugins.onAssembly(new FlowableFlatMapPublisher(sources, SingleInternalHelper.toFlowable(), false, Integer.MAX_VALUE, Flowable.bufferSize()));

}

/**

* Flattens a {@code Single} that emits a {@code Single} into a single {@code Single} that emits the item

* emitted by the nested {@code Single}, without any transformation.

* <p>

* <img width="640" height="370" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.merge.oo.png" alt="">

* <p>

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code merge} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param <T> the value type of the sources and the output

* @param source

* a {@code Single} that emits a {@code Single}

* @return a {@code Single} that emits the item that is the result of flattening the {@code Single} emitted

* by {@code source}

* @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings({ "unchecked", "rawtypes" })

public static <T> Single<T> merge(SingleSource<? extends SingleSource<? extends T>> source) {

ObjectHelper.requireNonNull(source, "source is null");

return RxJavaPlugins.onAssembly(new SingleFlatMap<SingleSource<? extends T>, T>(source, (Function)Functions.identity()));

}

/**

* Flattens two Singles into a single Flowable, without any transformation.

* <p>

* <img width="640" height="380" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.merge.png" alt="">

* <p>

* You can combine items emitted by multiple Singles so that they appear as a single Flowable, by

* using the {@code merge} method.

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code merge} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param <T> the common value type

* @param source1

* a Single to be merged

* @param source2

* a Single to be merged

* @return a Flowable that emits all of the items emitted by the source Singles

* @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> merge(

SingleSource<? extends T> source1, SingleSource<? extends T> source2

) {

ObjectHelper.requireNonNull(source1, "source1 is null");

ObjectHelper.requireNonNull(source2, "source2 is null");

return merge(Flowable.fromArray(source1, source2));

}

/**

* Flattens three Singles into a single Flowable, without any transformation.

* <p>

* <img width="640" height="380" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.merge.png" alt="">

* <p>

* You can combine items emitted by multiple Singles so that they appear as a single Flowable, by using

* the {@code merge} method.

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code merge} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param <T> the common value type

* @param source1

* a Single to be merged

* @param source2

* a Single to be merged

* @param source3

* a Single to be merged

* @return a Flowable that emits all of the items emitted by the source Singles

* @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> merge(

SingleSource<? extends T> source1, SingleSource<? extends T> source2,

SingleSource<? extends T> source3

) {

ObjectHelper.requireNonNull(source1, "source1 is null");

ObjectHelper.requireNonNull(source2, "source2 is null");

ObjectHelper.requireNonNull(source3, "source3 is null");

return merge(Flowable.fromArray(source1, source2, source3));

}

/**

* Flattens four Singles into a single Flowable, without any transformation.

* <p>

* <img width="640" height="380" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Single.merge.png" alt="">

* <p>

* You can combine items emitted by multiple Singles so that they appear as a single Flowable, by using

* the {@code merge} method.

* <dl>

* <dt><b>Backpressure:</b></dt>

* <dd>The returned {@code Flowable} honors the backpressure of the downstream consumer.</dd>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code merge} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

*

* @param <T> the common value type

* @param source1

* a Single to be merged

* @param source2

* a Single to be merged

* @param source3

* a Single to be merged

* @param source4

* a Single to be merged

* @return a Flowable that emits all of the items emitted by the source Singles

* @see <a href="http://reactivex.io/documentation/operators/merge.html">ReactiveX operators documentation: Merge</a>

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> merge(

SingleSource<? extends T> source1, SingleSource<? extends T> source2,

SingleSource<? extends T> source3, SingleSource<? extends T> source4

) {

ObjectHelper.requireNonNull(source1, "source1 is null");

ObjectHelper.requireNonNull(source2, "source2 is null");

ObjectHelper.requireNonNull(source3, "source3 is null");

ObjectHelper.requireNonNull(source4, "source4 is null");

return merge(Flowable.fromArray(source1, source2, source3, source4));

}

/**

* Returns a singleton instance of a never-signalling Single (only calls onSubscribe).

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code never} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the target value type

* @return the singleton never instance

* @since 2.0

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Single<T> never() {

return RxJavaPlugins.onAssembly((Single<T>) SingleNever.INSTANCE);

}

/**

* Signals success with 0L value after the given delay for each SingleObserver.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code timer} operates by default on the {@code computation} {@link Scheduler}.</dd>

* </dl>

* @param delay the delay amount

* @param unit the time unit of the delay

* @return the new Single instance

* @since 2.0

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.COMPUTATION)

public static Single<Long> timer(long delay, TimeUnit unit) {

return timer(delay, unit, Schedulers.computation());

}

/**

* Signals success with 0L value after the given delay for each SingleObserver.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>you specify the {@link Scheduler} to signal on.</dd>

* </dl>

* @param delay the delay amount

* @param unit the time unit of the delay

* @param scheduler the scheduler where the single 0L will be emitted

* @return the new Single instance

* @throws NullPointerException

* if unit is null, or

* if scheduler is null

* @since 2.0

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.CUSTOM)

public static Single<Long> timer(final long delay, final TimeUnit unit, final Scheduler scheduler) {

ObjectHelper.requireNonNull(unit, "unit is null");

ObjectHelper.requireNonNull(scheduler, "scheduler is null");

return RxJavaPlugins.onAssembly(new SingleTimer(delay, unit, scheduler));

}

/**

* Compares two SingleSources and emits true if they emit the same value (compared via Object.equals).

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code equals} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the common value type

* @param first the first SingleSource instance

* @param second the second SingleSource instance

* @return the new Single instance

* @since 2.0

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<Boolean> equals(final SingleSource<? extends T> first, final SingleSource<? extends T> second) { // NOPMD

ObjectHelper.requireNonNull(first, "first is null");

ObjectHelper.requireNonNull(second, "second is null");

return RxJavaPlugins.onAssembly(new SingleEquals<T>(first, second));

}

/**

* <strong>Advanced use only:</strong> creates a Single instance without

* any safeguards by using a callback that is called with a SingleObserver.

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code unsafeCreate} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type

* @param onSubscribe the function that is called with the subscribing SingleObserver

* @return the new Single instance

* @throws IllegalArgumentException if {@code source} is a subclass of {@code Single}; such

* instances don't need conversion and is possibly a port remnant from 1.x or one should use {@link #hide()}

* instead.

* @since 2.0

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Single<T> unsafeCreate(SingleSource<T> onSubscribe) {

ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");

if (onSubscribe instanceof Single) {

throw new IllegalArgumentException("unsafeCreate(Single) should be upgraded");

}

return RxJavaPlugins.onAssembly(new SingleFromUnsafeSource<T>(onSubscribe));

}

/**

* Allows using and disposing a resource while running a SingleSource instance generated from

* that resource (similar to a try-with-resources).

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code using} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type of the SingleSource generated

* @param <U> the resource type

* @param resourceSupplier the Callable called for each SingleObserver to generate a resource Object

* @param singleFunction the function called with the returned resource

* Object from {@code resourceSupplier} and should return a SingleSource instance

* to be run by the operator

* @param disposer the consumer of the generated resource that is called exactly once for

* that particular resource when the generated SingleSource terminates

* (successfully or with an error) or gets cancelled.

* @return the new Single instance

* @since 2.0

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T, U> Single<T> using(Callable<U> resourceSupplier,

Function<? super U, ? extends SingleSource<? extends T>> singleFunction,

Consumer<? super U> disposer) {

return using(resourceSupplier, singleFunction, disposer, true);

}

/**

* Allows using and disposing a resource while running a SingleSource instance generated from

* that resource (similar to a try-with-resources).

* <dl>

* <dt><b>Scheduler:</b></dt>

* <dd>{@code using} does not operate by default on a particular {@link Scheduler}.</dd>

* </dl>

* @param <T> the value type of the SingleSource generated

* @param <U> the resource type

* @param resourceSupplier the Callable called for each SingleObserver to generate a resource Object

* @param singleFunction the function called with the returned resource

* Object from {@code resourceSupplier} and should return a SingleSource instance

* to be run by the operator

* @param disposer the consumer of the generated resource that is called exactly once for

* that particular resource when the generated SingleSource terminates

* (successfully or with an error) or gets cancelled.

* @param eager

* if true, the disposer is called before the terminal event is signalled

* if false, the disposer is called after the terminal event is delivered to downstream

* @return the new Single instance

* @since 2.0

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T, U> Single<T> using(

final Callable<U> resourceSupplier,

final Function<? super U, ? extends SingleSource<? extends T>> singleFunction,

final Consumer<? super U> disposer,

final boolean eager) {

ObjectHelper.requireNonNull(resourceSupplier, "resourceSupplier is null");

ObjectHelper.requireNonNull(singleFunction, "singleFunction is null");

ObjectHelper.requireNonNull(disposer, "disposer is null");

return RxJavaPlugins.onAssembly(new SingleUsing<T, U>(resourceSupplier, singleFunction, disposer, eager));

}

/**

* Wraps a SingleSource instance into a new Single instance if not already a Single

* instance.

* <dl>