OperatorSwitch.java 文件源码

public SwitchSubscriber(Subscriber<? super T> child) {
    s = new SerializedSubscriber<T>(child);
    ssub = new SerialSubscription();
    child.add(ssub);
    child.setProducer(new Producer(){

        @Override
        public void request(long n) {
            if (infinite) {
                return;
            }
            if(n == Long.MAX_VALUE) {
                infinite = true;
            }
            InnerSubscriber localSubscriber;
            synchronized (guard) {
                localSubscriber = currentSubscriber;
                if (currentSubscriber == null) {
                    initialRequested = n;
                } else {
                    // If n == Long.MAX_VALUE, infinite will become true. Then currentSubscriber.requested won't be used.
                    // Therefore we don't need to worry about overflow.
                    currentSubscriber.requested += n;
                }
            }
            if (localSubscriber != null) {
                localSubscriber.requestMore(n);
            }
        }
    });
}