PendingResultObservable.java 文件源码

@Override
public void call(final Subscriber<? super T> subscriber) {
    result.setResultCallback(new ResultCallback<T>() {
        @Override
        public void onResult(T t) {
            subscriber.onNext(t);
            complete = true;
            subscriber.onCompleted();
        }
    });
    subscriber.add(Subscriptions.create(new Action0() {
        @Override
        public void call() {
            if (!complete) {
                result.cancel();
            }
        }
    }));
}