OperatorTimeoutWithSelector.java 文件源码

public Subscription call(final TimeoutSubscriber<T> timeoutSubscriber, final Long seqId, T value, Worker inner) {
    try {
        return ((Observable) this.val$timeoutSelector.call(value)).unsafeSubscribe(new Subscriber<V>() {
            public void onCompleted() {
                timeoutSubscriber.onTimeout(seqId.longValue());
            }

            public void onError(Throwable e) {
                timeoutSubscriber.onError(e);
            }

            public void onNext(V v) {
                timeoutSubscriber.onTimeout(seqId.longValue());
            }
        });
    } catch (Throwable t) {
        Exceptions.throwOrReport(t, timeoutSubscriber);
        return Subscriptions.unsubscribed();
    }
}