ContinueWithIfEmptyOperator.java

package com.trello.core.rx;

import rx.Observable;
import rx.Subscriber;

/**
 * Passes control to another Observable if the source observable does not emit any items.
 *
 * Handy if you've a CacheObservable that does a cheap lookup and a DoHardWorkObservable that does the
 * hard work and you want to only do the hard work if CacheObservable is empty.
 *
 * You'd do something like: CacheObservable.lift(new ContinueWithOperator(DoHardWorkObservable)).
 */
public class ContinueWithIfEmptyOperator<T> implements Observable.Operator<T, T> {

    private final Observable<? extends T> mResumeObservable;

    public ContinueWithIfEmptyOperator(Observable<? extends T> resumeObservable) {
        mResumeObservable = resumeObservable;
    }

    @Override
    public Subscriber<? super T> call(final Subscriber<? super T> child) {
        Subscriber<T> s = new Subscriber<T>() {
            boolean mHasEmitted;

            @Override
            public void onNext(T t) {
                if (!child.isUnsubscribed()) {
                    mHasEmitted = true;
                    child.onNext(t);
                }
            }

            @Override
            public void onError(Throwable e) {
                if (!child.isUnsubscribed()) {
                    child.onError(e);
                }
            }

            @Override
            public void onCompleted() {
                if (child.isUnsubscribed()) {
                    return;
                }

                if (mHasEmitted) {
                    child.onCompleted();
                    return;
                }

                // Nothing emitted, so we subscribe to the resume observable
                unsubscribe();
                mResumeObservable.unsafeSubscribe(child);
            }
        };

        child.add(s);

        return s;
    }
}