OperatorTimeoutBase.java 文件源码

public Subscriber<? super T> call(Subscriber<? super T> subscriber) {
    Worker inner = this.scheduler.createWorker();
    subscriber.add(inner);
    SerialSubscription serial = new SerialSubscription();
    subscriber.add(serial);
    TimeoutSubscriber<T> timeoutSubscriber = new TimeoutSubscriber(new SerializedSubscriber(subscriber), this.timeoutStub, serial, this.other, inner);
    serial.set((Subscription) this.firstTimeoutStub.call(timeoutSubscriber, Long.valueOf(0), inner));
    return timeoutSubscriber;
}