Completable.java

/**
 * 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 elapsed.
     * @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 elapsed
     * 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) {
                                    e = new CompositeException(Arrays.asList(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];