/**

* Copyright 2014 Netflix, Inc.

*

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

import java.util.*;

import java.util.concurrent.*;

import java.util.concurrent.atomic.AtomicBoolean;

import rx.Observable.OnSubscribe;

import rx.annotations.Experimental;

import rx.exceptions.*;

import rx.functions.*;

import rx.internal.operators.*;

import rx.internal.util.*;

import rx.plugins.*;

import rx.schedulers.Schedulers;

import rx.subscriptions.*;

/**

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

*/

@Experimental

public class Completable {

/**

* Callback used for building deferred computations that takes a CompletableSubscriber.

*/

public interface CompletableOnSubscribe extends Action1<CompletableSubscriber> {

}

/**

* Convenience interface and callback used by the lift operator that given a child CompletableSubscriber,

* return a parent CompletableSubscriber that does any kind of lifecycle-related transformations.

*/

public interface CompletableOperator extends Func1<CompletableSubscriber, CompletableSubscriber> {

}

/**

* Represents the subscription API callbacks when subscribing to a Completable instance.

*/

public interface CompletableSubscriber {

/**

* Called once the deferred computation completes normally.

*/

void onCompleted();

/**

* Called once if the deferred computation 'throws' an exception.

* @param e the exception, not null.

*/

void onError(Throwable e);

/**

* Called once by the Completable to set a Subscription on this instance which

* then can be used to cancel the subscription at any time.

* @param d the Subscription instance to call dispose on for cancellation, not null

*/

void onSubscribe(Subscription d);

}

/**

* Convenience interface and callback used by the compose operator to turn a Completable into another

* Completable fluently.

*/

public interface CompletableTransformer extends Func1<Completable, Completable> {

}

/** Single instance of a complete Completable. */

static final Completable COMPLETE = create(new CompletableOnSubscribe() {

@Override

public void call(CompletableSubscriber s) {

s.onSubscribe(Subscriptions.unsubscribed());

s.onCompleted();

}

});

/** Single instance of a never Completable. */

static final Completable NEVER = create(new CompletableOnSubscribe() {

@Override

public void call(CompletableSubscriber s) {

s.onSubscribe(Subscriptions.unsubscribed());

}

});

/** The error handler instance. */

static final RxJavaErrorHandler ERROR_HANDLER = RxJavaPlugins.getInstance().getErrorHandler();

/**

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

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

* @param sources the array of source Completables

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

public static Completable amb(final Completable... sources) {

requireNonNull(sources);

if (sources.length == 0) {

return complete();

}

if (sources.length == 1) {

return sources[0];

}

return create(new CompletableOnSubscribe() {

@Override

public void call(final CompletableSubscriber s) {

final CompositeSubscription set = new CompositeSubscription();

s.onSubscribe(set);

final AtomicBoolean once = new AtomicBoolean();

CompletableSubscriber inner = new CompletableSubscriber() {

@Override

public void onCompleted() {

if (once.compareAndSet(false, true)) {

set.unsubscribe();

s.onCompleted();

}

}

@Override

public void onError(Throwable e) {

if (once.compareAndSet(false, true)) {

set.unsubscribe();

s.onError(e);

} else {

ERROR_HANDLER.handleError(e);

}

}

@Override

public void onSubscribe(Subscription d) {

set.add(d);

}

};

for (Completable c : sources) {

if (set.isUnsubscribed()) {

return;

}

if (c == null) {

NullPointerException npe = new NullPointerException("One of the sources is null");

if (once.compareAndSet(false, true)) {

set.unsubscribe();

s.onError(npe);

} else {

ERROR_HANDLER.handleError(npe);

}

return;

}

if (once.get() || set.isUnsubscribed()) {

return;

}

// no need to have separate subscribers because inner is stateless

c.subscribe(inner);

}

}

});

}

/**

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

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

* @param sources the array of source Completables

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

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

requireNonNull(sources);

return create(new CompletableOnSubscribe() {

@Override

public void call(final CompletableSubscriber s) {

final CompositeSubscription set = new CompositeSubscription();

s.onSubscribe(set);

final AtomicBoolean once = new AtomicBoolean();

CompletableSubscriber inner = new CompletableSubscriber() {

@Override

public void onCompleted() {

if (once.compareAndSet(false, true)) {

set.unsubscribe();

s.onCompleted();

}

}

@Override

public void onError(Throwable e) {

if (once.compareAndSet(false, true)) {

set.unsubscribe();

s.onError(e);

} else {

ERROR_HANDLER.handleError(e);

}

}

@Override

public void onSubscribe(Subscription d) {

set.add(d);

}

};

Iterator<? extends Completable> it;

try {

it = sources.iterator();

} catch (Throwable e) {

s.onError(e);

return;

}

if (it == null) {

s.onError(new NullPointerException("The iterator returned is null"));

return;

}

boolean empty = true;

for (;;) {

if (once.get() || set.isUnsubscribed()) {

return;

}

boolean b;

try {

b = it.hasNext();

} catch (Throwable e) {

if (once.compareAndSet(false, true)) {

set.unsubscribe();

s.onError(e);

} else {

ERROR_HANDLER.handleError(e);

}

return;

}

if (!b) {

if (empty) {

s.onCompleted();

}

break;

}

empty = false;

if (once.get() || set.isUnsubscribed()) {

return;

}

Completable c;

try {

c = it.next();

} catch (Throwable e) {

if (once.compareAndSet(false, true)) {

set.unsubscribe();

s.onError(e);

} else {

ERROR_HANDLER.handleError(e);

}

return;

}

if (c == null) {

NullPointerException npe = new NullPointerException("One of the sources is null");

if (once.compareAndSet(false, true)) {

set.unsubscribe();

s.onError(npe);

} else {

ERROR_HANDLER.handleError(npe);

}

return;

}

if (once.get() || set.isUnsubscribed()) {

return;

}

// no need to have separate subscribers because inner is stateless

c.subscribe(inner);

}

}

});

}

/**

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

* @return a Completable instance that completes immediately

*/

public static Completable complete() {

return COMPLETE;

}

/**

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

* @param sources the sources to concatenate

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

* @throws NullPointerException if sources is null

*/

public static Completable concat(Completable... sources) {

requireNonNull(sources);

if (sources.length == 0) {

return complete();

} else

if (sources.length == 1) {

return sources[0];

}

return create(new CompletableOnSubscribeConcatArray(sources));

}

/**

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

* @param sources the sources to concatenate

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

* @throws NullPointerException if sources is null

*/

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

requireNonNull(sources);

return create(new CompletableOnSubscribeConcatIterable(sources));

}

/**

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

* @param sources the sources to concatenate

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

* @throws NullPointerException if sources is null

*/

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

return concat(sources, 2);

}

/**

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

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

*/

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

requireNonNull(sources);

if (prefetch < 1) {

throw new IllegalArgumentException("prefetch > 0 required but it was " + prefetch);

}

return create(new CompletableOnSubscribeConcat(sources, prefetch));

}

/**

* Constructs a Completable instance by wrapping the given onSubscribe callback.

* @param onSubscribe the callback which will receive the CompletableSubscriber instances

* when the Completable is subscribed to.

* @return the created Completable instance

* @throws NullPointerException if onSubscribe is null

*/

public static Completable create(CompletableOnSubscribe onSubscribe) {

requireNonNull(onSubscribe);

try {

// TODO plugin wrapping onSubscribe

return new Completable(onSubscribe);

} catch (NullPointerException ex) {

throw ex;

} catch (Throwable ex) {

ERROR_HANDLER.handleError(ex);

throw toNpe(ex);

}

}

/**

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

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

* @return the Completable instance

*/

public static Completable defer(final Func0<? extends Completable> completableFunc0) {

requireNonNull(completableFunc0);

return create(new CompletableOnSubscribe() {

@Override

public void call(CompletableSubscriber s) {

Completable c;

try {

c = completableFunc0.call();

} catch (Throwable e) {

s.onSubscribe(Subscriptions.unsubscribed());

s.onError(e);

return;

}

if (c == null) {

s.onSubscribe(Subscriptions.unsubscribed());

s.onError(new NullPointerException("The completable returned is null"));

return;

}

c.subscribe(s);

}

});

}

/**

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

* and emits its returned Throwable.

* <p>

* If the errorFunc0 returns null, the child CompletableSubscribers will receive a

* NullPointerException.

* @param errorFunc0 the error supplier, not null

* @return the new Completable instance

* @throws NullPointerException if errorFunc0 is null

*/

public static Completable error(final Func0<? extends Throwable> errorFunc0) {

requireNonNull(errorFunc0);

return create(new CompletableOnSubscribe() {

@Override

public void call(CompletableSubscriber s) {

s.onSubscribe(Subscriptions.unsubscribed());

Throwable error;

try {

error = errorFunc0.call();

} catch (Throwable e) {

error = e;

}

if (error == null) {

error = new NullPointerException("The error supplied is null");

}

s.onError(error);

}

});

}

/**

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

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

* @return the new Completable instance

* @throws NullPointerException if error is null

*/

public static Completable error(final Throwable error) {

requireNonNull(error);

return create(new CompletableOnSubscribe() {

@Override

public void call(CompletableSubscriber s) {

s.onSubscribe(Subscriptions.unsubscribed());

s.onError(error);

}

});

}

/**

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

* emits either an unchecked exception or simply completes.

* @param run the runnable to run for each subscriber

* @return the new Completable instance

* @throws NullPointerException if run is null

*/

public static Completable fromAction(final Action0 action) {

requireNonNull(action);

return create(new CompletableOnSubscribe() {

@Override

public void call(CompletableSubscriber s) {

BooleanSubscription bs = new BooleanSubscription();

s.onSubscribe(bs);

try {

action.call();

} catch (Throwable e) {

if (!bs.isUnsubscribed()) {

s.onError(e);

}

return;

}

if (!bs.isUnsubscribed()) {

s.onCompleted();

}

}

});

}

/**

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

* normal result and emits onError or onCompleted only.

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

* @return the new Completable instance

*/

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

requireNonNull(callable);

return create(new CompletableOnSubscribe() {

@Override

public void call(CompletableSubscriber s) {

BooleanSubscription bs = new BooleanSubscription();

s.onSubscribe(bs);

try {

callable.call();

} catch (Throwable e) {

if (!bs.isUnsubscribed()) {

s.onError(e);

}

return;

}

if (!bs.isUnsubscribed()) {

s.onCompleted();

}

}

});

}

/**

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

* @param future the future to react to

* @return the new Completable instance

*/

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

requireNonNull(future);

return fromObservable(Observable.from(future));

}

/**

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

* emits only the terminal event.

* @param flowable the Flowable instance to subscribe to, not null

* @return the new Completable instance

* @throws NullPointerException if flowable is null

*/

public static Completable fromObservable(final Observable<?> flowable) {

requireNonNull(flowable);

return create(new CompletableOnSubscribe() {

@Override

public void call(final CompletableSubscriber cs) {

Subscriber<Object> subscriber = new Subscriber<Object>() {

@Override

public void onCompleted() {

cs.onCompleted();

}

@Override

public void onError(Throwable t) {

cs.onError(t);

}

@Override

public void onNext(Object t) {

// ignored

}

};

cs.onSubscribe(subscriber);

flowable.unsafeSubscribe(subscriber);

}

});

}

/**

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

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

* @return the new Completable instance

* @throws NullPointerException if single is null

*/

public static Completable fromSingle(final Single<?> single) {

requireNonNull(single);

return create(new CompletableOnSubscribe() {

@Override

public void call(final CompletableSubscriber s) {

SingleSubscriber<Object> te = new SingleSubscriber<Object>() {

@Override

public void onError(Throwable e) {

s.onError(e);

}

@Override

public void onSuccess(Object value) {

s.onCompleted();

}

};

s.onSubscribe(te);

single.subscribe(te);

}

});

}

/**

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

* @param sources the iterable sequence of sources.

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

public static Completable merge(Completable... sources) {

requireNonNull(sources);

if (sources.length == 0) {

return complete();

} else

if (sources.length == 1) {

return sources[0];

}

return create(new CompletableOnSubscribeMergeArray(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.

* @param sources the iterable sequence of sources.

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

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

requireNonNull(sources);

return create(new CompletableOnSubscribeMergeIterable(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.

* @param sources the iterable sequence of sources.

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

public static Completable merge(Observable<? extends Completable> 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.

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

*/

public static Completable merge(Observable<? extends Completable> 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.

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

*/

protected static Completable merge0(Observable<? extends Completable> sources, int maxConcurrency, boolean delayErrors) {

requireNonNull(sources);

if (maxConcurrency < 1) {

throw new IllegalArgumentException("maxConcurrency > 0 required but it was " + maxConcurrency);

}

return create(new CompletableOnSubscribeMerge(sources, maxConcurrency, delayErrors));

}

/**

* Returns a Completable 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.

* @param sources the array of Completables

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

public static Completable mergeDelayError(Completable... sources) {

requireNonNull(sources);

return create(new CompletableOnSubscribeMergeDelayErrorArray(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.

* @param sources the sequence of Completables

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

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

requireNonNull(sources);

return create(new CompletableOnSubscribeMergeDelayErrorIterable(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.

* @param sources the sequence of Completables

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

public static Completable mergeDelayError(Observable<? extends Completable> 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.

* @param sources the sequence of Completables

* @return the new Completable instance

* @throws NullPointerException if sources is null

*/

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

return merge0(sources, maxConcurrency, true);

}

/**

* Returns a Completable that never calls onError or onComplete.

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

*/

public static Completable never() {

return NEVER;

}

/**

* Java 7 backport: throws a NullPointerException if o is null.

* @param o the object to check

* @return the o value

* @throws NullPointerException if o is null

*/

static <T> T requireNonNull(T o) {

if (o == null) {

throw new NullPointerException();

}

return o;

}

/**

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

* @param delay the delay time

* @param unit the delay unit

* @return the new Completable instance

*/

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 ellapsed

* by using the supplied scheduler.

* @param delay the delay time

* @param unit the delay unit

* @return the new Completable instance

*/

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

requireNonNull(unit);

requireNonNull(scheduler);

return create(new CompletableOnSubscribe() {

@Override

public void call(final CompletableSubscriber s) {

MultipleAssignmentSubscription mad = new MultipleAssignmentSubscription();

s.onSubscribe(mad);

if (!mad.isUnsubscribed()) {

final Scheduler.Worker w = scheduler.createWorker();

mad.set(w);

w.schedule(new Action0() {

@Override

public void call() {

try {

s.onCompleted();

} finally {

w.unsubscribe();

}

}

}, delay, unit);

}

}

});

}

/**

* 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

*/

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 performs an eager unsubscription before the terminal event is emitted.

*

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

* @param completableFunc1 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

*/

public static <R> Completable using(Func0<R> resourceFunc0,

Func1<? super R, ? extends Completable> completableFunc1,

Action1<? super R> disposer) {

return using(resourceFunc0, completableFunc1, 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 unsubscription after the terminal event is emitted.

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

*

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

* @param completableFunc1 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

*/

public static <R> Completable using(final Func0<R> resourceFunc0,

final Func1<? super R, ? extends Completable> completableFunc1,

final Action1<? super R> disposer,

final boolean eager) {

requireNonNull(resourceFunc0);

requireNonNull(completableFunc1);

requireNonNull(disposer);

return create(new CompletableOnSubscribe() {

@Override

public void call(final CompletableSubscriber s) {

final R resource;

try {

resource = resourceFunc0.call();

} catch (Throwable e) {

s.onSubscribe(Subscriptions.unsubscribed());

s.onError(e);

return;

}

Completable cs;

try {

cs = completableFunc1.call(resource);

} catch (Throwable e) {

try {

disposer.call(resource);

} catch (Throwable ex) {

Exceptions.throwIfFatal(e);

Exceptions.throwIfFatal(ex);

s.onSubscribe(Subscriptions.unsubscribed());

s.onError(new CompositeException(Arrays.asList(e, ex)));

return;

}

Exceptions.throwIfFatal(e);

s.onSubscribe(Subscriptions.unsubscribed());

s.onError(e);

return;

}

if (cs == null) {

try {

disposer.call(resource);

} catch (Throwable ex) {

Exceptions.throwIfFatal(ex);

s.onSubscribe(Subscriptions.unsubscribed());

s.onError(new CompositeException(Arrays.asList(new NullPointerException("The completable supplied is null"), ex)));

return;

}

s.onSubscribe(Subscriptions.unsubscribed());

s.onError(new NullPointerException("The completable supplied is null"));

return;

}

final AtomicBoolean once = new AtomicBoolean();

cs.subscribe(new CompletableSubscriber() {

Subscription d;

void dispose() {

d.unsubscribe();

if (once.compareAndSet(false, true)) {

try {

disposer.call(resource);

} catch (Throwable ex) {

ERROR_HANDLER.handleError(ex);

}

}

}

@Override

public void onCompleted() {

if (eager) {

if (once.compareAndSet(false, true)) {

try {

disposer.call(resource);

} catch (Throwable ex) {

s.onError(ex);

return;

}

}

}

s.onCompleted();

if (!eager) {

dispose();

}

}

@Override

public void onError(Throwable e) {

if (eager) {

if (once.compareAndSet(false, true)) {

try {

disposer.call(resource);

} catch (Throwable ex) {

ex.addSuppressed(e);

e = ex;

}

}

}

s.onError(e);

if (!eager) {

dispose();

}

}

@Override

public void onSubscribe(Subscription d) {

this.d = d;

s.onSubscribe(Subscriptions.create(new Action0() {

@Override

public void call() {

dispose();

}

}));

}

});

}

});

}

/** The actual subscription action. */

private final CompletableOnSubscribe onSubscribe;

/**

* Constructs a Completable instance with the given onSubscribe callback.

* @param onSubscribe the callback that will receive CompletableSubscribers when they subscribe,

* not null (not verified)

*/

protected Completable(CompletableOnSubscribe onSubscribe) {

this.onSubscribe = onSubscribe;

}

/**

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

* or the other Completable whichever fires first.

* @param other the other Completable, not null

* @return the new Completable instance

* @throws NullPointerException if other is null

*/

public final Completable ambWith(Completable other) {

requireNonNull(other);

return amb(this, other);

}

/**

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

* rethrows any exception emitted.

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

*/

public final void await() {

final CountDownLatch cdl = new CountDownLatch(1);

final Throwable[] err = new Throwable[1];