/**

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

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.BlockingMultiObserver;

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

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

import io.reactivex.internal.util.*;

import io.reactivex.observers.TestObserver;

import io.reactivex.plugins.RxJavaPlugins;

import io.reactivex.schedulers.Schedulers;

/**

* Represents a deferred computation and emission of a maybe value or exception.

* <p>

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

* <p>

* The main consumer type of Maybe is {@link MaybeObserver} whose methods are called

* in a sequential fashion following this protocol:<br>

* {@code onSubscribe (onSuccess | onError | onComplete)?}.

* <p>

* @param <T> the value type

* @since 2.0

*/

public abstract class Maybe<T> implements MaybeSource<T> {

/**

* Runs multiple Maybe 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 the Iterable.

* @return the new Maybe instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Maybe<T> amb(final Iterable<? extends MaybeSource<? extends T>> sources) {

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

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

}

/**

* Runs multiple Maybe 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 the array.

* @return the new Maybe instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Maybe<T> ambArray(final MaybeSource<? extends T>... sources) {

if (sources.length == 0) {

return empty();

}

if (sources.length == 1) {

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

}

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

}

/**

* Concatenate the single values, in a non-overlapping fashion, of the MaybeSource 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 MaybeSource instances

* @return the new Flowable instance

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

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

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

return RxJavaPlugins.onAssembly(new MaybeConcatIterable<T>(sources));

}

/**

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

* <p>

* <img width="640" height="422" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.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 MaybeSource to be concatenated

* @param source2

* a MaybeSource to be concatenated

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

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

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> concat(MaybeSource<? extends T> source1, MaybeSource<? extends T> source2) {

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

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

return concatArray(source1, source2);

}

/**

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

* <p>

* <img width="640" height="422" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.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 MaybeSource to be concatenated

* @param source2

* a MaybeSource to be concatenated

* @param source3

* a MaybeSource to be concatenated

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

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

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> concat(

MaybeSource<? extends T> source1, MaybeSource<? extends T> source2, MaybeSource<? extends T> source3) {

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

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

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

return concatArray(source1, source2, source3);

}

/**

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

* <p>

* <img width="640" height="422" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.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 MaybeSource to be concatenated

* @param source2

* a MaybeSource to be concatenated

* @param source3

* a MaybeSource to be concatenated

* @param source4

* a MaybeSource to be concatenated

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

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

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> concat(

MaybeSource<? extends T> source1, MaybeSource<? extends T> source2, MaybeSource<? extends T> source3, MaybeSource<? 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 concatArray(source1, source2, source3, source4);

}

/**

* Concatenate the single values, in a non-overlapping fashion, of the MaybeSource 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

* expects the {@code Publisher} to honor backpressure as well. If the sources {@code Publisher}

* violates this, a {@link io.reactivex.exceptions.MissingBackpressureException} is signalled.</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 MaybeSource instances

* @return the new Flowable instance

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

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

return concat(sources, 2);

}

/**

* Concatenate the single values, in a non-overlapping fashion, of the MaybeSource 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

* expects the {@code Publisher} to honor backpressure as well. If the sources {@code Publisher}

* violates this, a {@link io.reactivex.exceptions.MissingBackpressureException} is signalled.</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 MaybeSource instances

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

* @return the new Flowable instance

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

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

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

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

ObjectHelper.verifyPositive(prefetch, "prefetch");

return RxJavaPlugins.onAssembly(new FlowableConcatMapPublisher(sources, MaybeToPublisher.instance(), prefetch, ErrorMode.IMMEDIATE));

}

/**

* Concatenate the single values, in a non-overlapping fashion, of the MaybeSource sources in the 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 MaybeSource instances

* @return the new Flowable instance

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

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

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

if (sources.length == 0) {

return Flowable.empty();

}

if (sources.length == 1) {

return RxJavaPlugins.onAssembly(new MaybeToFlowable<T>((MaybeSource<T>)sources[0]));

}

return RxJavaPlugins.onAssembly(new MaybeConcatArray<T>(sources));

}

/**

* Concatenates a variable number of MaybeSource sources and delays errors from any of them

* till all terminate.

* <p>

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

* <dl>

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

* <dd>The operator honors backpressure from downstream.</dd>

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

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

* </dl>

* @param sources the array of sources

* @param <T> the common base value type

* @return the new Flowable instance

* @throws NullPointerException if sources is null

*/

@SuppressWarnings("unchecked")

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Flowable<T> concatArrayDelayError(MaybeSource<? extends T>... sources) {

if (sources.length == 0) {

return Flowable.empty();

} else

if (sources.length == 1) {

return RxJavaPlugins.onAssembly(new MaybeToFlowable<T>((MaybeSource<T>)sources[0]));

}

return RxJavaPlugins.onAssembly(new MaybeConcatArrayDelayError<T>(sources));

}

/**

* Concatenates a sequence of MaybeSource eagerly into a single stream of values.

* <p>

* Eager concatenation means that once a subscriber subscribes, this operator subscribes to all of the

* source MaybeSources. The operator buffers the value emitted by these MaybeSources and then drains them

* in order, each one after the previous one completes.

* <dl>

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

* <dd>The operator honors backpressure from downstream.</dd>

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

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

* </dl>

* @param <T> the value type

* @param sources a sequence of MaybeSources that need to be eagerly concatenated

* @return the new Flowable instance with the specified concatenation behavior

*/

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

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Flowable<T> concatArrayEager(MaybeSource<? extends T>... sources) {

return Flowable.fromArray(sources).concatMapEager((Function)MaybeToPublisher.instance());

}

/**

* Concatenates the Iterable sequence of MaybeSources into a single sequence by subscribing to each MaybeSource,

* one after the other, one at a time and delays any errors till the all inner MaybeSources terminate.

*

* <dl>

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

* <dd>The operator honors backpressure from downstream.</dd>

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

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

* </dl>

*

* @param <T> the common element base type

* @param sources the Iterable sequence of MaybeSources

* @return the new Flowable with the concatenating behavior

*/

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

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Flowable<T> concatDelayError(Iterable<? extends MaybeSource<? extends T>> sources) {

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

return Flowable.fromIterable(sources).concatMapDelayError((Function)MaybeToPublisher.instance());

}

/**

* Concatenates the Publisher sequence of Publishers into a single sequence by subscribing to each inner Publisher,

* one after the other, one at a time and delays any errors till the all inner and the outer Publishers terminate.

*

* <dl>

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

* <dd>{@code concatDelayError} fully supports backpressure.</dd>

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

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

* </dl>

*

* @param <T> the common element base type

* @param sources the Publisher sequence of Publishers

* @return the new Publisher with the concatenating behavior

*/

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

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Flowable<T> concatDelayError(Publisher<? extends MaybeSource<? extends T>> sources) {

return Flowable.fromPublisher(sources).concatMapDelayError((Function)MaybeToPublisher.instance());

}

/**

* Concatenates a sequence of MaybeSources eagerly into a single stream of values.

* <p>

* Eager concatenation means that once a subscriber subscribes, this operator subscribes to all of the

* source MaybeSources. The operator buffers the values emitted by these MaybeSources and then drains them

* in order, each one after the previous one completes.

* <dl>

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

* <dd>Backpressure is honored towards the downstream.</dd>

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

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

* </dl>

* @param <T> the value type

* @param sources a sequence of MaybeSource that need to be eagerly concatenated

* @return the new Flowable instance with the specified concatenation behavior

*/

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

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Flowable<T> concatEager(Iterable<? extends MaybeSource<? extends T>> sources) {

return Flowable.fromIterable(sources).concatMapEager((Function)MaybeToPublisher.instance());

}

/**

* Concatenates a Publisher sequence of Publishers eagerly into a single stream of values.

* <p>

* Eager concatenation means that once a subscriber subscribes, this operator subscribes to all of the

* emitted source Publishers as they are observed. The operator buffers the values emitted by these

* Publishers and then drains them in order, each one after the previous one completes.

* <dl>

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

* <dd>Backpressure is honored towards the downstream and the outer Publisher is

* expected to support backpressure. Violating this assumption, the operator will

* signal {@link io.reactivex.exceptions.MissingBackpressureException}.</dd>

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

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

* </dl>

* @param <T> the value type

* @param sources a sequence of Publishers that need to be eagerly concatenated

* @return the new Publisher instance with the specified concatenation behavior

*/

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

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Flowable<T> concatEager(Publisher<? extends MaybeSource<? extends T>> sources) {

return Flowable.fromPublisher(sources).concatMapEager((Function)MaybeToPublisher.instance());

}

/**

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

* <p>

* Example:

* <pre><code>

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

* Callback listener = new Callback() {

* &#64;Override

* public void onEvent(Event e) {

* if (e.isNothing()) {

* emitter.onComplete();

* } else {

* 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 onSubscribe the emitter that is called when a MaybeObserver subscribes to the returned {@code Maybe}

* @return the new Maybe instance

* @see MaybeOnSubscribe

* @see Cancellable

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Maybe<T> create(MaybeOnSubscribe<T> onSubscribe) {

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

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

}

/**

* Calls a Callable for each individual MaybeObserver to return the actual MaybeSource 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 maybeSupplier the Callable that is called for each individual MaybeObserver and

* returns a MaybeSource instance to subscribe to

* @return the new Maybe instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Maybe<T> defer(final Callable<? extends MaybeSource<? extends T>> maybeSupplier) {

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

return RxJavaPlugins.onAssembly(new MaybeDefer<T>(maybeSupplier));

}

/**

* Returns a (singleton) Maybe instance that calls {@link MaybeObserver#onComplete onComplete}

* immediately.

* <p>

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

* <dl>

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

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

* </dl>

* @param <T> the value type

* @return the new Maybe instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Maybe<T> empty() {

return RxJavaPlugins.onAssembly((Maybe<T>)MaybeEmpty.INSTANCE);

}

/**

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

* subscriber subscribes to it.

* <p>

* <img width="640" height="447" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.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 MaybeObserver#onError onError}

* @param <T>

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

* @return a Maybe that invokes the subscriber's {@link MaybeObserver#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> Maybe<T> error(Throwable exception) {

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

return RxJavaPlugins.onAssembly(new MaybeError<T>(exception));

}

/**

* Returns a Maybe that invokes a {@link MaybeObserver}'s {@link MaybeObserver#onError onError} method when the

* MaybeObserver subscribes to it.

* <p>

* <img width="640" height="190" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/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 supplier

* a Callable factory to return a Throwable for each individual MaybeObserver

* @param <T>

* the type of the items (ostensibly) emitted by the Maybe

* @return a Maybe that invokes the {@link MaybeObserver}'s {@link MaybeObserver#onError onError} method when

* the MaybeObserver 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> Maybe<T> error(Callable<? extends Throwable> supplier) {

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

return RxJavaPlugins.onAssembly(new MaybeErrorCallable<T>(supplier));

}

/**

* Returns a Maybe instance that runs the given Action for each subscriber and

* emits either its exception or simply completes.

* <dl>

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

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

* </dl>

* @param <T> the target type

* @param run the runnable to run for each subscriber

* @return the new Maybe instance

* @throws NullPointerException if run is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Maybe<T> fromAction(final Action run) {

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

return RxJavaPlugins.onAssembly(new MaybeFromAction<T>(run));

}

/**

* Wraps a CompletableSource into a Maybe.

*

* <dl>

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

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

* </dl>

* @param <T> the target type

* @param completableSource the CompletableSource to convert from

* @return the new Maybe instance

* @throws NullPointerException if completable is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Maybe<T> fromCompletable(CompletableSource completableSource) {

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

return RxJavaPlugins.onAssembly(new MaybeFromCompletable<T>(completableSource));

}

/**

* Wraps a SingleSource into a Maybe.

*

* <dl>

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

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

* </dl>

* @param <T> the target type

* @param singleSource the SingleSource to convert from

* @return the new Maybe instance

* @throws NullPointerException if single is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Maybe<T> fromSingle(SingleSource<T> singleSource) {

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

return RxJavaPlugins.onAssembly(new MaybeFromSingle<T>(singleSource));

}

/**

* Returns a {@link Maybe} that invokes passed function and emits its result for each new MaybeObserver that subscribes

* while considering {@code null} value from the callable as indication for valueless completion.

* <p>

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

* It makes passed function "lazy".

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

* <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 MaybeObserver will subscribe to the {@link Maybe}.

* @param <T>

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

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

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

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

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

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

}

/**

* Converts a {@link Future} into a Maybe, treating a null result as an indication of emptiness.

* <p>

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

* <p>

* You can convert any object that supports the {@link Future} interface into a Maybe 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 Maybe is blocking; you cannot unsubscribe from it.

* <p>

* Unlike 1.x, cancelling the Maybe won't cancel the future. If necessary, one can use composition to achieve the

* cancellation effect: {@code futureMaybe.doOnDispose(() -> future.cancel(true));}.

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

* @return a Maybe 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> Maybe<T> fromFuture(Future<? extends T> future) {

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

return RxJavaPlugins.onAssembly(new MaybeFromFuture<T>(future, 0L, null));

}

/**

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

* <p>

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

* <p>

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

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

* method.

* <p>

* Unlike 1.x, cancelling the Maybe won't cancel the future. If necessary, one can use composition to achieve the

* cancellation effect: {@code futureMaybe.doOnCancel(() -> future.cancel(true));}.

* <p>

* <em>Important note:</em> This Maybe is blocking on the thread it gets subscribed on; you cannot unsubscribe from 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 Maybe

* @return a Maybe 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> Maybe<T> fromFuture(Future<? extends T> future, long timeout, TimeUnit unit) {

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

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

return RxJavaPlugins.onAssembly(new MaybeFromFuture<T>(future, timeout, unit));

}

/**

* Returns a Maybe instance that runs the given Action for each subscriber and

* emits either its exception or simply completes.

* <dl>

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

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

* </dl>

* @param <T> the target type

* @param run the runnable to run for each subscriber

* @return the new Maybe instance

* @throws NullPointerException if run is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Maybe<T> fromRunnable(final Runnable run) {

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

return RxJavaPlugins.onAssembly(new MaybeFromRunnable<T>(run));

}

/**

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

* <p>

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

* <p>

* To convert any object into a {@code Maybe} 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 Maybe} 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> Maybe<T> just(T item) {

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

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

}

/**

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

* running all MaybeSources at once.

* <dl>

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

* <dd>The operator honors backpressure from downstream.</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 MaybeSource sources

* @return the new Flowable instance

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

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

return merge(Flowable.fromIterable(sources));

}

/**

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

* running all MaybeSources at once.

* <dl>

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

* <dd>The operator honors backpressure from downstream.</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 MaybeSource sources

* @return the new Flowable instance

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

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

return merge(sources, Integer.MAX_VALUE);

}

/**

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

* running at most maxConcurrency MaybeSources at once.

* <dl>

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

* <dd>The operator honors backpressure from downstream.</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 MaybeSource sources

* @param maxConcurrency the maximum number of concurrently running MaybeSources

* @return the new Flowable instance

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

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

public static <T> Flowable<T> merge(Publisher<? extends MaybeSource<? extends T>> sources, int maxConcurrency) {

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

ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");

return RxJavaPlugins.onAssembly(new FlowableFlatMapPublisher(sources, MaybeToPublisher.instance(), false, maxConcurrency, Flowable.bufferSize()));

}

/**

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

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

* <p>

* <img width="640" height="393" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.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 MaybeSource} that emits a {@code MaybeSource}

* @return a {@code Maybe} that emits the item that is the result of flattening the {@code MaybeSource} 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> Maybe<T> merge(MaybeSource<? extends MaybeSource<? extends T>> source) {

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

return RxJavaPlugins.onAssembly(new MaybeFlatten(source, Functions.identity()));

}

/**

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

* <p>

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

* <p>

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

* using the {@code merge} method.

* <dl>

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

* <dd>The operator honors backpressure from downstream.</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 MaybeSource to be merged

* @param source2

* a MaybeSource to be merged

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

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

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> merge(

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

) {

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

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

return mergeArray(source1, source2);

}

/**

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

* <p>

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

* <p>

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

* the {@code merge} method.

* <dl>

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

* <dd>The operator honors backpressure from downstream.</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 MaybeSource to be merged

* @param source2

* a MaybeSource to be merged

* @param source3

* a MaybeSource to be merged

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

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

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> merge(

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

MaybeSource<? extends T> source3

) {

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

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

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

return mergeArray(source1, source2, source3);

}

/**

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

* <p>

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

* <p>

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

* the {@code merge} method.

* <dl>

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

* <dd>The operator honors backpressure from downstream.</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 MaybeSource to be merged

* @param source2

* a MaybeSource to be merged

* @param source3

* a MaybeSource to be merged

* @param source4

* a MaybeSource to be merged

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

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

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> merge(

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

MaybeSource<? extends T> source3, MaybeSource<? 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 mergeArray(source1, source2, source3, source4);

}

/**

* Merges an array sequence of MaybeSource instances into a single Flowable sequence,

* running all MaybeSources at once.

* <dl>

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

* <dd>The operator honors backpressure from downstream.</dd>

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

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

* </dl>

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

* @param sources the array sequence of MaybeSource sources

* @return the new Flowable instance

*/

@BackpressureSupport(BackpressureKind.FULL)

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@SuppressWarnings("unchecked")

public static <T> Flowable<T> mergeArray(MaybeSource<? extends T>... sources) {

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

if (sources.length == 0) {

return Flowable.empty();

}

if (sources.length == 1) {

return RxJavaPlugins.onAssembly(new MaybeToFlowable<T>((MaybeSource<T>)sources[0]));

}

return RxJavaPlugins.onAssembly(new MaybeMergeArray<T>(sources));

}