RxJava2FutureUtils.java 文件源码

@Override
protected void subscribeActual(SingleObserver<? super T> observer) {
    ValueSourceDisposable disposable = new ValueSourceDisposable();
    valueSource.addCallbacks(
            result -> {
                try {
                    observer.onSuccess(result);
                } catch (Throwable e) {
                    observer.onError(e);
                }
            },
            ex -> {
                if (!disposable.isDisposed()) {
                    observer.onError(ex);
                }
            }
    );
    observer.onSubscribe(disposable);
}