/**

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

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

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

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

import io.reactivex.internal.util.ExceptionHelper;

import io.reactivex.observers.TestObserver;

import io.reactivex.plugins.RxJavaPlugins;

import io.reactivex.schedulers.Schedulers;

/**

* Represents a deferred computation without any value but only indication for completion or exception.

*

* The class follows a similar event pattern as Reactive-Streams: onSubscribe (onError|onComplete)?

*/

public abstract class Completable implements CompletableSource {

/**

* Returns a Completable which terminates as soon as one of the source Completables

* terminates (normally or with an error) and cancels all other Completables.

* <dl>

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

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

* </dl>

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

* occur in the same order as in this array.

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable ambArray(final CompletableSource... sources) {

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

if (sources.length == 0) {

return complete();

}

if (sources.length == 1) {

return wrap(sources[0]);

}

return RxJavaPlugins.onAssembly(new CompletableAmb(sources, null));

}

/**

* Returns a Completable which terminates as soon as one of the source Completables

* terminates (normally or with an error) and cancels all other Completables.

* <dl>

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

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

* </dl>

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

* occur in the same order as in this Iterable.

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable amb(final Iterable<? extends CompletableSource> sources) {

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

return RxJavaPlugins.onAssembly(new CompletableAmb(null, sources));

}

/**

* Returns a Completable instance that completes immediately when subscribed to.

* <dl>

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

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

* </dl>

* @return a Completable instance that completes immediately

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable complete() {

return RxJavaPlugins.onAssembly(CompletableEmpty.INSTANCE);

}

/**

* Returns a Completable which completes only when all sources complete, one after another.

* <dl>

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

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

* </dl>

* @param sources the sources to concatenate

* @return the Completable instance which completes only when all sources complete

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable concatArray(CompletableSource... sources) {

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

if (sources.length == 0) {

return complete();

} else

if (sources.length == 1) {

return wrap(sources[0]);

}

return RxJavaPlugins.onAssembly(new CompletableConcatArray(sources));

}

/**

* Returns a Completable which completes only when all sources complete, one after another.

* <dl>

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

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

* </dl>

* @param sources the sources to concatenate

* @return the Completable instance which completes only when all sources complete

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable concat(Iterable<? extends CompletableSource> sources) {

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

return RxJavaPlugins.onAssembly(new CompletableConcatIterable(sources));

}

/**

* Returns a Completable which completes only when all sources complete, one after another.

* <dl>

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

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

* and expects the other {@code Publisher} 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 sources the sources to concatenate

* @return the Completable instance which completes only when all sources complete

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@BackpressureSupport(BackpressureKind.FULL)

public static Completable concat(Publisher<? extends CompletableSource> sources) {

return concat(sources, 2);

}

/**

* Returns a Completable which completes only when all sources complete, one after another.

* <dl>

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

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

* and expects the other {@code Publisher} 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 sources the sources to concatenate

* @param prefetch the number of sources to prefetch from the sources

* @return the Completable instance which completes only when all sources complete

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@BackpressureSupport(BackpressureKind.FULL)

public static Completable concat(Publisher<? extends CompletableSource> sources, int prefetch) {

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

ObjectHelper.verifyPositive(prefetch, "prefetch");

return RxJavaPlugins.onAssembly(new CompletableConcat(sources, prefetch));

}

/**

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

* <p>

* Example:

* <pre><code>

* Completable.create(emitter -&gt; {

* Callback listener = new Callback() {

* &#64;Override

* public void onEvent(Event e) {

* emitter.onComplete();

* }

*

* &#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 source the emitter that is called when a CompletableObserver subscribes to the returned {@code Completable}

* @return the new Completable instance

* @see CompletableOnSubscribe

* @see Cancellable

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable create(CompletableOnSubscribe source) {

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

return RxJavaPlugins.onAssembly(new CompletableCreate(source));

}

/**

* Constructs a Completable instance by wrapping the given source callback

* <strong>without any safeguards; you should manage the lifecycle and response

* to downstream cancellation/dispose</strong>.

* <dl>

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

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

* </dl>

* @param source the callback which will receive the CompletableObserver instances

* when the Completable is subscribed to.

* @return the created Completable instance

* @throws NullPointerException if source is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable unsafeCreate(CompletableSource source) {

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

if (source instanceof Completable) {

throw new IllegalArgumentException("Use of unsafeCreate(Completable)!");

}

return RxJavaPlugins.onAssembly(new CompletableFromUnsafeSource(source));

}

/**

* Defers the subscription to a Completable instance returned by a supplier.

* <dl>

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

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

* </dl>

* @param completableSupplier the supplier that returns the Completable that will be subscribed to.

* @return the Completable instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable defer(final Callable<? extends CompletableSource> completableSupplier) {

ObjectHelper.requireNonNull(completableSupplier, "completableSupplier");

return RxJavaPlugins.onAssembly(new CompletableDefer(completableSupplier));

}

/**

* Creates a Completable which calls the given error supplier for each subscriber

* and emits its returned Throwable.

* <p>

* If the errorSupplier returns null, the child CompletableObservers will receive a

* NullPointerException.

* <dl>

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

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

* </dl>

* @param errorSupplier the error supplier, not null

* @return the new Completable instance

* @throws NullPointerException if errorSupplier is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

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

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

return RxJavaPlugins.onAssembly(new CompletableErrorSupplier(errorSupplier));

}

/**

* Creates a Completable instance that emits the given Throwable exception to subscribers.

* <dl>

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

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

* </dl>

* @param error the Throwable instance to emit, not null

* @return the new Completable instance

* @throws NullPointerException if error is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable error(final Throwable error) {

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

return RxJavaPlugins.onAssembly(new CompletableError(error));

}

/**

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

* emits either an unchecked 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 run the runnable to run for each subscriber

* @return the new Completable instance

* @throws NullPointerException if run is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable fromAction(final Action run) {

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

return RxJavaPlugins.onAssembly(new CompletableFromAction(run));

}

/**

* Returns a Completable which when subscribed, executes the callable function, ignores its

* normal result and emits onError or onComplete only.

* <dl>

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

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

* </dl>

* @param callable the callable instance to execute for each subscriber

* @return the new Completable instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable fromCallable(final Callable<?> callable) {

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

return RxJavaPlugins.onAssembly(new CompletableFromCallable(callable));

}

/**

* Returns a Completable instance that reacts to the termination of the given Future in a blocking fashion.

* <p>

* Note that cancellation from any of the subscribers to this Completable will cancel the future.

* <dl>

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

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

* </dl>

* @param future the future to react to

* @return the new Completable instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable fromFuture(final Future<?> future) {

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

return fromAction(Functions.futureAction(future));

}

/**

* Returns a Completable instance that runs the given Runnable 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 run the runnable to run for each subscriber

* @return the new Completable instance

* @throws NullPointerException if run is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable fromRunnable(final Runnable run) {

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

return RxJavaPlugins.onAssembly(new CompletableFromRunnable(run));

}

/**

* Returns a Completable instance that subscribes to the given Observable, ignores all values and

* emits only the terminal event.

* <dl>

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

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

* </dl>

* @param <T> the type of the Observable

* @param observable the Observable instance to subscribe to, not null

* @return the new Completable instance

* @throws NullPointerException if flowable is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Completable fromObservable(final ObservableSource<T> observable) {

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

return RxJavaPlugins.onAssembly(new CompletableFromObservable<T>(observable));

}

/**

* Returns a Completable instance that subscribes to the given publisher, ignores all values and

* emits only the terminal event.

* <dl>

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

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

* and expects the other {@code Publisher} to honor it as well.</dd>

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

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

* </dl>

* @param <T> the type of the publisher

* @param publisher the Publisher instance to subscribe to, not null

* @return the new Completable instance

* @throws NullPointerException if publisher is null

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Completable fromPublisher(final Publisher<T> publisher) {

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

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

}

/**

* Returns a Completable instance that when subscribed to, subscribes to the Single instance and

* emits a completion event if the single emits onSuccess or forwards any onError events.

* <dl>

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

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

* </dl>

* @param <T> the value type of the Single

* @param single the Single instance to subscribe to, not null

* @return the new Completable instance

* @throws NullPointerException if single is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <T> Completable fromSingle(final SingleSource<T> single) {

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

return RxJavaPlugins.onAssembly(new CompletableFromSingle<T>(single));

}

/**

* Returns a Completable instance that subscribes to all sources at once and

* completes only when all source Completables complete or one of them emits an error.

* <dl>

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

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

* </dl>

* @param sources the iterable sequence of sources.

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable mergeArray(CompletableSource... sources) {

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

if (sources.length == 0) {

return complete();

} else

if (sources.length == 1) {

return wrap(sources[0]);

}

return RxJavaPlugins.onAssembly(new CompletableMergeArray(sources));

}

/**

* Returns a Completable instance that subscribes to all sources at once and

* completes only when all source Completables complete or one of them emits an error.

* <dl>

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

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

* </dl>

* @param sources the iterable sequence of sources.

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable merge(Iterable<? extends CompletableSource> sources) {

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

return RxJavaPlugins.onAssembly(new CompletableMergeIterable(sources));

}

/**

* Returns a Completable instance that subscribes to all sources at once and

* completes only when all source Completables complete or one of them emits an error.

* <dl>

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

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

* and expects the other {@code Publisher} to honor it as well.</dd>

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

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

* </dl>

* @param sources the iterable sequence of sources.

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)

public static Completable merge(Publisher<? extends CompletableSource> sources) {

return merge0(sources, Integer.MAX_VALUE, false);

}

/**

* Returns a Completable instance that keeps subscriptions to a limited number of sources at once and

* completes only when all source Completables complete or one of them emits an error.

* <dl>

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

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

* and expects the other {@code Publisher} to honor it as well.</dd>

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

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

* </dl>

* @param sources the iterable sequence of sources.

* @param maxConcurrency the maximum number of concurrent subscriptions

* @return the new Completable instance

* @throws NullPointerException if sources is null

* @throws IllegalArgumentException if maxConcurrency is less than 1

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@BackpressureSupport(BackpressureKind.FULL)

public static Completable merge(Publisher<? extends CompletableSource> sources, int maxConcurrency) {

return merge0(sources, maxConcurrency, false);

}

/**

* Returns a Completable instance that keeps subscriptions to a limited number of sources at once and

* completes only when all source Completables terminate in one way or another, combining any exceptions

* thrown by either the sources Observable or the inner Completable instances.

* <dl>

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

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

* and expects the other {@code Publisher} to honor it as well.

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

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

* </dl>

* @param sources the iterable sequence of sources.

* @param maxConcurrency the maximum number of concurrent subscriptions

* @param delayErrors delay all errors from the main source and from the inner Completables?

* @return the new Completable instance

* @throws NullPointerException if sources is null

* @throws IllegalArgumentException if maxConcurrency is less than 1

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@BackpressureSupport(BackpressureKind.FULL)

private static Completable merge0(Publisher<? extends CompletableSource> sources, int maxConcurrency, boolean delayErrors) {

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

ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency");

return RxJavaPlugins.onAssembly(new CompletableMerge(sources, maxConcurrency, delayErrors));

}

/**

* Returns a CompletableConsumable that subscribes to all Completables in the source array and delays

* any error emitted by either the sources observable or any of the inner Completables until all of

* them terminate in a way or another.

* <dl>

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

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

* </dl>

* @param sources the array of Completables

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable mergeArrayDelayError(CompletableSource... sources) {

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

return RxJavaPlugins.onAssembly(new CompletableMergeDelayErrorArray(sources));

}

/**

* Returns a Completable that subscribes to all Completables in the source sequence and delays

* any error emitted by either the sources observable or any of the inner Completables until all of

* them terminate in a way or another.

* <dl>

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

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

* </dl>

* @param sources the sequence of Completables

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable mergeDelayError(Iterable<? extends CompletableSource> sources) {

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

return RxJavaPlugins.onAssembly(new CompletableMergeDelayErrorIterable(sources));

}

/**

* Returns a Completable that subscribes to all Completables in the source sequence and delays

* any error emitted by either the sources observable or any of the inner Completables until all of

* them terminate in a way or another.

* <dl>

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

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

* and expects the other {@code Publisher} to honor it as well.</dd>

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

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

* </dl>

* @param sources the sequence of Completables

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)

public static Completable mergeDelayError(Publisher<? extends CompletableSource> sources) {

return merge0(sources, Integer.MAX_VALUE, true);

}

/**

* Returns a Completable that subscribes to a limited number of inner Completables at once in

* the source sequence and delays any error emitted by either the sources

* observable or any of the inner Completables until all of

* them terminate in a way or another.

* <dl>

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

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

* and expects the other {@code Publisher} to honor it as well.</dd>

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

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

* </dl>

* @param sources the sequence of Completables

* @param maxConcurrency the maximum number of concurrent subscriptions to Completables

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@BackpressureSupport(BackpressureKind.FULL)

public static Completable mergeDelayError(Publisher<? extends CompletableSource> sources, int maxConcurrency) {

return merge0(sources, maxConcurrency, true);

}

/**

* Returns a Completable that never calls onError or onComplete.

* <dl>

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

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

* </dl>

* @return the singleton instance that never calls onError or onComplete

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable never() {

return RxJavaPlugins.onAssembly(CompletableNever.INSTANCE);

}

/**

* Returns a Completable instance that fires its onComplete event after the given delay elapsed.

* <dl>

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

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

* </dl>

* @param delay the delay time

* @param unit the delay unit

* @return the new Completable instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.COMPUTATION)

public static Completable timer(long delay, TimeUnit unit) {

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

}

/**

* Returns a Completable instance that fires its onComplete event after the given delay elapsed

* by using the supplied scheduler.

* <dl>

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

* <dd>{@code timer} operates on the {@link Scheduler} you specify.</dd>

* </dl>

* @param delay the delay time

* @param unit the delay unit

* @param scheduler the scheduler where to emit the complete event

* @return the new Completable instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.CUSTOM)

public static Completable 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 CompletableTimer(delay, unit, scheduler));

}

/**

* Creates a NullPointerException instance and sets the given Throwable as its initial cause.

* @param ex the Throwable instance to use as cause, not null (not verified)

* @return the created NullPointerException

*/

private static NullPointerException toNpe(Throwable ex) {

NullPointerException npe = new NullPointerException("Actually not, but can't pass out an exception otherwise...");

npe.initCause(ex);

return npe;

}

/**

* Returns a Completable instance which manages a resource along

* with a custom Completable instance while the subscription is active.

* <p>

* This overload disposes eagerly before the terminal event is emitted.

* <dl>

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

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

* </dl>

* @param <R> the resource type

* @param resourceSupplier the supplier that returns a resource to be managed.

* @param completableFunction the function that given a resource returns a Completable instance that will be subscribed to

* @param disposer the consumer that disposes the resource created by the resource supplier

* @return the new Completable instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <R> Completable using(Callable<R> resourceSupplier,

Function<? super R, ? extends CompletableSource> completableFunction,

Consumer<? super R> disposer) {

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

}

/**

* Returns a Completable instance which manages a resource along

* with a custom Completable instance while the subscription is active and performs eager or lazy

* resource disposition.

* <p>

* If this overload performs a lazy cancellation after the terminal event is emitted.

* Exceptions thrown at this time will be delivered to RxJavaPlugins only.

* <dl>

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

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

* </dl>

* @param <R> the resource type

* @param resourceSupplier the supplier that returns a resource to be managed

* @param completableFunction the function that given a resource returns a non-null

* Completable instance that will be subscribed to

* @param disposer the consumer that disposes the resource created by the resource supplier

* @param eager if true, the resource is disposed before the terminal event is emitted, if false, the

* resource is disposed after the terminal event has been emitted

* @return the new Completable instance

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static <R> Completable using(

final Callable<R> resourceSupplier,

final Function<? super R, ? extends CompletableSource> completableFunction,

final Consumer<? super R> disposer,

final boolean eager) {

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

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

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

return RxJavaPlugins.onAssembly(new CompletableUsing<R>(resourceSupplier, completableFunction, disposer, eager));

}

/**

* Wraps the given CompletableSource into a Completable

* if not already Completable.

* <dl>

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

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

* </dl>

* @param source the source to wrap

* @return the source or its wrapper Completable

* @throws NullPointerException if source is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public static Completable wrap(CompletableSource source) {

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

if (source instanceof Completable) {

return RxJavaPlugins.onAssembly((Completable)source);

}

return RxJavaPlugins.onAssembly(new CompletableFromUnsafeSource(source));

}

/**

* Returns a Completable that emits the a terminated event of either this Completable

* or the other Completable whichever fires first.

* <dl>

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

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

* </dl>

* @param other the other Completable, not null. A subscription to this provided source will occur after subscribing

* to the current source.

* @return the new Completable instance

* @throws NullPointerException if other is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public final Completable ambWith(CompletableSource other) {

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

return ambArray(this, other);

}

/**

* Returns an Observable which will subscribe to this Completable and once that is completed then

* will subscribe to the {@code next} ObservableSource. An error event from this Completable will be

* propagated to the downstream subscriber and will result in skipping the subscription of the

* Observable.

* <dl>

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

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

* </dl>

* @param <T> the value type of the next ObservableSource

* @param next the Observable to subscribe after this Completable is completed, not null

* @return Observable that composes this Completable and next

* @throws NullPointerException if next is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public final <T> Observable<T> andThen(ObservableSource<T> next) {

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

return RxJavaPlugins.onAssembly(new ObservableDelaySubscriptionOther<T, Object>(next, toObservable()));

}

/**

* Returns a Flowable which will subscribe to this Completable and once that is completed then

* will subscribe to the {@code next} Flowable. An error event from this Completable will be

* propagated to the downstream subscriber and will result in skipping the subscription of the

* Publisher.

* <dl>

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

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

* and expects the other {@code Publisher} to honor it as well.</dd>

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

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

* </dl>

* @param <T> the value type of the next Publisher

* @param next the Publisher to subscribe after this Completable is completed, not null

* @return Flowable that composes this Completable and next

* @throws NullPointerException if next is null

*/

@CheckReturnValue

@BackpressureSupport(BackpressureKind.FULL)

@SchedulerSupport(SchedulerSupport.NONE)

public final <T> Flowable<T> andThen(Publisher<T> next) {

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

return RxJavaPlugins.onAssembly(new FlowableDelaySubscriptionOther<T, Object>(next, toFlowable()));

}

/**

* Returns a Single which will subscribe to this Completable and once that is completed then

* will subscribe to the {@code next} SingleSource. An error event from this Completable will be

* propagated to the downstream subscriber and will result in skipping the subscription of the

* Single.

* <dl>

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

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

* </dl>

*

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

* @param next the Single to subscribe after this Completable is completed, not null

* @return Single that composes this Completable and next

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public final <T> Single<T> andThen(SingleSource<T> next) {

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

return RxJavaPlugins.onAssembly(new SingleDelayWithCompletable<T>(next, this));

}

/**

* Returns a {@link Maybe} which will subscribe to this Completable and once that is completed then

* will subscribe to the {@code next} MaybeSource. An error event from this Completable will be

* propagated to the downstream subscriber and will result in skipping the subscription of the

* Maybe.

* <dl>

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

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

* </dl>

*

* @param <T> the value type of the next MaybeSource

* @param next the Maybe to subscribe after this Completable is completed, not null

* @return Maybe that composes this Completable and next

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public final <T> Maybe<T> andThen(MaybeSource<T> next) {

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

return RxJavaPlugins.onAssembly(new MaybeDelayWithCompletable<T>(next, this));

}

/**

* Returns a Completable that first runs this Completable

* and then the other completable.

* <p>

* This is an alias for {@link #concatWith(CompletableSource)}.

* <dl>

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

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

* </dl>

* @param next the other Completable, not null

* @return the new Completable instance

* @throws NullPointerException if other is null

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public final Completable andThen(CompletableSource next) {

return concatWith(next);

}

/**

* Subscribes to and awaits the termination of this Completable instance in a blocking manner and

* rethrows any exception emitted.

* <dl>

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

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

* </dl>

* @throws RuntimeException wrapping an InterruptedException if the current thread is interrupted

*/

@SchedulerSupport(SchedulerSupport.NONE)

public final void blockingAwait() {

BlockingMultiObserver<Void> observer = new BlockingMultiObserver<Void>();

subscribe(observer);

observer.blockingGet();

}

/**

* Subscribes to and awaits the termination of this Completable instance in a blocking manner

* with a specific timeout and rethrows any exception emitted within the timeout window.

* <dl>

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

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

* </dl>

* @param timeout the timeout value

* @param unit the timeout unit

* @return true if the this Completable instance completed normally within the time limit,

* false if the timeout elapsed before this Completable terminated.

* @throws RuntimeException wrapping an InterruptedException if the current thread is interrupted

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public final boolean blockingAwait(long timeout, TimeUnit unit) {

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

BlockingMultiObserver<Void> observer = new BlockingMultiObserver<Void>();

subscribe(observer);

return observer.blockingAwait(timeout, unit);

}

/**

* Subscribes to this Completable instance and blocks until it terminates, then returns null or

* the emitted exception if any.

* <dl>

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

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

* </dl>

* @return the throwable if this terminated with an error, null otherwise

* @throws RuntimeException that wraps an InterruptedException if the wait is interrupted

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public final Throwable blockingGet() {

BlockingMultiObserver<Void> observer = new BlockingMultiObserver<Void>();

subscribe(observer);

return observer.blockingGetError();

}

/**

* Subscribes to this Completable instance and blocks until it terminates or the specified timeout

* elapses, then returns null for normal termination or the emitted exception if any.

* <dl>

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

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

* </dl>

* @param timeout the timeout value

* @param unit the time unit

* @return the throwable if this terminated with an error, null otherwise

* @throws RuntimeException that wraps an InterruptedException if the wait is interrupted or

* TimeoutException if the specified timeout elapsed before it

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

public final Throwable blockingGet(long timeout, TimeUnit unit) {

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

BlockingMultiObserver<Void> observer = new BlockingMultiObserver<Void>();

subscribe(observer);

return observer.blockingGetError(timeout, unit);

}

/**

* Subscribes to this Completable only once, when the first CompletableObserver

* subscribes to the result Completable, caches its terminal event

* and relays/replays it to observers.

* <p>

* Note that this operator doesn't allow disposing the connection

* of the upstream source.

* <dl>

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

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

* </dl>

* @return the new Completable instance

* @since 2.0.4 - experimental

*/

@CheckReturnValue

@SchedulerSupport(SchedulerSupport.NONE)

@Experimental

public final Completable cache() {

return RxJavaPlugins.onAssembly(new CompletableCache(this));

}

/**

* Calls the given transformer function with this instance and returns the function's resulting