Java 类io.reactivex.processors.AsyncProcessor 实例源码

/**
 * Create a Flowable that signals the terminal value or error of the given
 * CompletionStage.
 * <p>Cancelling the Flowable subscription doesn't cancel the CompletionStage.
 * @param <T> the value type
 * @param cs the CompletionStage instance
 * @return the new Flowable instance
 */
public static <T> Flowable<T> fromFuture(CompletionStage<T> cs) {
    AsyncProcessor<T> ap = AsyncProcessor.create();
    cs.whenComplete((v, e) -> {
        if (e != null) {
            ap.onError(e);
        } else {
            ap.onNext(v);
            ap.onComplete();
        }
    });
    return ap;
}

@Override
public void getAppSettings() {
    mSplashProcessor = AsyncProcessor.create();
    mSplashDisposable = mSplashProcessor.subscribeWith(new SplashSubscriber());
    new SplashRequestManager(deviceId).
            getAppConfig().
            subscribe(mSplashProcessor);
}

FlowableCacheLast(Publisher<T> source) {
    this.source = source;
    this.processor = AsyncProcessor.create();
    this.once = new AtomicBoolean();
}