/**

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

import io.reactivex.internal.util.*;

import io.reactivex.observers.TestObserver;

import io.reactivex.plugins.RxJavaPlugins;

import io.reactivex.schedulers.Schedulers;

/**

* The {@code Maybe} class represents a deferred computation and emission of a single value, no value at all or an exception.

* <p>

* The {@code Maybe} class implements the {@link MaybeSource} base interface and the default consumer

* type it interacts with is the {@link MaybeObserver} via the {@link #subscribe(MaybeObserver)} method.

* <p>

* The {@code Maybe} operates with the following sequential protocol:

* <pre><code>

* onSubscribe (onSuccess | onError | onComplete)?

* </code></pre>

* <p>

* Note that {@code onSuccess}, {@code onError} and {@code onComplete} are mutually exclusive events; unlike {@code Observable},

* {@code onSuccess} is never followed by {@code onError} or {@code onComplete}.

* <p>

* Like {@link Observable}, a running {@code Maybe} can be stopped through the {@link Disposable} instance

* provided to consumers through {@link MaybeObserver#onSubscribe}.

* <p>

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

* asynchronous. {@code Maybe} instances returned by the methods of this class are <em>cold</em>

* and there is a standard <em>hot</em> implementation in the form of a subject:

* {@link io.reactivex.subjects.MaybeSubject MaybeSubject}.

* <p>

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

* <p>

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

* <p>

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

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

* <p>

* Example:

* <pre><code>

* Disposable d = Maybe.just("Hello World")

* .delay(10, TimeUnit.SECONDS, Schedulers.io())

* .subscribeWith(new DisposableMaybeObserver&lt;String&gt;() {

* &#64;Override

* public void onStart() {

* System.out.println("Started");

* }

*

* &#64;Override

* public void onSuccess(String value) {

* System.out.println("Success: " + value);

* }

*

* &#64;Override

* public void onError(Throwable error) {

* error.printStackTrace();

* }

*

* &#64;Override

* public void onComplete() {

* System.out.println("Done!");

* }

* });

*

* Thread.sleep(5000);

*

* d.dispose();

* </code></pre>

* <p>

* Note that by design, subscriptions via {@link #subscribe(MaybeObserver)} can't be disposed

* from the outside (hence the

* {@code void} return of the {@link #subscribe(MaybeObserver)} method) and it is the

* responsibility of the implementor of the {@code MaybeObserver} to allow this to happen.

* RxJava supports such usage with the standard

* {@link io.reactivex.observers.DisposableMaybeObserver DisposableMaybeObserver} instance.

* For convenience, the {@link #subscribeWith(MaybeObserver)} method is provided as well to

* allow working with a {@code MaybeObserver} (or subclass) instance to be applied with in

* a fluent manner (such as in the example above).

*

* @param <T> the value type

* @since 2.0

* @see io.reactivex.observers.DisposableMaybeObserver

*/

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

/**

* Runs multiple MaybeSources and signals the events of the first one that signals (disposing

* the rest).

* <p>

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

* <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 MaybeSources and signals the events of the first one that signals (disposing

* the rest).

* <p>

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

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

* <p>

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

* <p>

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

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

* <p>

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

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

* <p>

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

* <p>

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

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

* <p>

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

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

* <p>

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

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

* <p>

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

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

* <p>

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

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

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

* <dt><b>Error handling:</b></dt>

* <dd> If the {@link Action} throws an exception, the respective {@link Throwable} is

* delivered to the downstream via {@link MaybeObserver#onError(Throwable)},

* except when the downstream has disposed this {@code Maybe} source.

* In this latter case, the {@code Throwable} is delivered to the global error handler via

* {@link RxJavaPlugins#onError(Throwable)} as an {@link io.reactivex.exceptions.UndeliverableException UndeliverableException}.

* </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 the given {@link Callable} for each individual {@link MaybeObserver} that

* subscribes and emits the resulting non-null item via {@code onSuccess} while

* considering a {@code null} result from the {@code Callable} as indication for valueless completion

* via {@code onComplete}.

* <p>

* This operator allows you to defer the execution of the given {@code Callable} until a {@code MaybeObserver}

* subscribes to the returned {@link Maybe}. In other terms, this source operator evaluates the given

* {@code Callable} "lazily".

* <p>

* Note that the {@code null} handling of this operator differs from the similar source operators in the other

* {@link io.reactivex base reactive classes}. Those operators signal a {@code NullPointerException} if the value returned by their

* {@code Callable} is {@code null} while this {@code fromCallable} considers it to indicate the

* returned {@code Maybe} is empty.

* <dl>

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

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

* <dt><b>Error handling:</b></dt>

* <dd>Any non-fatal exception thrown by {@link Callable#call()} will be forwarded to {@code onError},

* except if the {@code MaybeObserver} disposed the subscription in the meantime. In this latter case,

* the exception is forwarded to the global error handler via

* {@link io.reactivex.plugins.RxJavaPlugins#onError(Throwable)} wrapped into a

* {@link io.reactivex.exceptions.UndeliverableException UndeliverableException}.

* Fatal exceptions are rethrown and usually will end up in the executing thread's

* {@link java.lang.Thread.UncaughtExceptionHandler#uncaughtException(Thread, Throwable)} handler.</dd>

* </dl>

*

* @param callable

* a {@link Callable} instance whose execution should be deferred and performed for each individual

* {@code MaybeObserver} that subscribes to the returned {@link Maybe}.

* @param <T>

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

* @return a new Maybe instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Maybe<T> fromCallable(@NonNull 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 dispose it.

* <p>

* Unlike 1.x, disposing 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, disposing 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 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 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>

* <dt><b>Error handling:</b></dt>

* <dd>If any of the source {@code MaybeSource}s signal a {@code Throwable} via {@code onError}, the resulting

* {@code Flowable} terminates with that {@code Throwable} and all other source {@code MaybeSource}s are disposed.

* If more than one {@code MaybeSource} signals an error, the resulting {@code Flowable} may terminate with the

* first one's error or, depending on the concurrency of the sources, may terminate with a

* {@code CompositeException} containing two or more of the various error signals.

* {@code Throwable}s that didn't make into the composite will be sent (individually) to the global error handler via

* {@link RxJavaPlugins#onError(Throwable)} method as {@code UndeliverableException} errors. Similarly, {@code Throwable}s

* signaled by source(s) after the returned {@code Flowable} has been cancelled or terminated with a

* (composite) error will be sent to the same global error handler.

* Use {@link #mergeDelayError(Iterable)} to merge sources and terminate only when all source {@code MaybeSource}s

* have completed or failed with an error.

* </dd>

* </dl>

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

* @param sources the Iterable sequence of MaybeSource sources

* @return the new Flowable instance

* @see #mergeDelayError(Iterable)

*/

@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>

* <dt><b>Error handling:</b></dt>

* <dd>If any of the source {@code MaybeSource}s signal a {@code Throwable} via {@code onError}, the resulting

* {@code Flowable} terminates with that {@code Throwable} and all other source {@code MaybeSource}s are disposed.

* If more than one {@code MaybeSource} signals an error, the resulting {@code Flowable} may terminate with the

* first one's error or, depending on the concurrency of the sources, may terminate with a

* {@code CompositeException} containing two or more of the various error signals.

* {@code Throwable}s that didn't make into the composite will be sent (individually) to the global error handler via

* {@link RxJavaPlugins#onError(Throwable)} method as {@code UndeliverableException} errors. Similarly, {@code Throwable}s

* signaled by source(s) after the returned {@code Flowable} has been cancelled or terminated with a

* (composite) error will be sent to the same global error handler.

* Use {@link #mergeDelayError(Publisher)} to merge sources and terminate only when all source {@code MaybeSource}s

* have completed or failed with an error.

* </dd>

* </dl>

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

* @param sources the Flowable sequence of MaybeSource sources

* @return the new Flowable instance

* @see #mergeDelayError(Publisher)

*/

@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>

* <dt><b>Error handling:</b></dt>

* <dd>If any of the source {@code MaybeSource}s signal a {@code Throwable} via {@code onError}, the resulting

* {@code Flowable} terminates with that {@code Throwable} and all other source {@code MaybeSource}s are disposed.

* If more than one {@code MaybeSource} signals an error, the resulting {@code Flowable} may terminate with the

* first one's error or, depending on the concurrency of the sources, may terminate with a

* {@code CompositeException} containing two or more of the various error signals.

* {@code Throwable}s that didn't make into the composite will be sent (individually) to the global error handler via

* {@link RxJavaPlugins#onError(Throwable)} method as {@code UndeliverableException} errors. Similarly, {@code Throwable}s

* signaled by source(s) after the returned {@code Flowable} has been cancelled or terminated with a

* (composite) error will be sent to the same global error handler.

* Use {@link #mergeDelayError(Publisher, int)} to merge sources and terminate only when all source {@code MaybeSource}s

* have completed or failed with an error.

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

* @see #mergeDelayError(Publisher, int)

*/

@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, 1));

}

/**

* 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="">

* <dl>

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

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

* <dt><b>Error handling:</b></dt>

* <dd>The resulting {@code Maybe} emits the outer source's or the inner {@code MaybeSource}'s {@code Throwable} as is.

* Unlike the other {@code merge()} operators, this operator won't and can't produce a {@code CompositeException} because there is

* only one possibility for the outer or the inner {@code MaybeSource} to emit an {@code onError} signal.

* Therefore, there is no need for a {@code mergeDelayError(MaybeSource<MaybeSource<T>>)} operator.

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