java类rx.subscriptions.SerialSubscription的实例源码

/**
 * Executes the interceptor chain for the passed request and response.
 *
 * @param request Request to be executed.
 * @param response Response to be populated.
 * @param keyEvaluationContext The context for {@link InterceptorKey} evaluation.
 *
 * @return The final result of execution after executing all the inbound and outbound interceptors and the router.
 */
public Observable<Void> execute(final I request, final O response, C keyEvaluationContext) {
    final ExecutionContext context = new ExecutionContext(request, keyEvaluationContext);
    InboundInterceptor<I, O> nextIn = context.nextIn(request);
    Observable<Void> startingPoint;

    if (null != nextIn) {
        startingPoint = nextIn.in(request, response);
    } else if (context.invokeRouter()){
        startingPoint = router.handle(request, response);
    } else {
        return Observable.error(new IllegalStateException("No router defined.")); // No router defined.
    }

    return startingPoint.lift(new Observable.Operator<Void, Void>() {
        @Override
        public Subscriber<? super Void> call(Subscriber<? super Void> child) {
            SerialSubscription subscription = new SerialSubscription();
            ChainSubscriber chainSubscriber = new ChainSubscriber(subscription, context, request, response, child);
            subscription.set(chainSubscriber);
            child.add(subscription);
            return chainSubscriber;
        }
    });
}

public SourceSubscriber(Subscriber<? super T> child, Func2<Integer, Throwable, Boolean> predicate, Worker inner, SerialSubscription serialSubscription, ProducerArbiter pa) {
    this.child = child;
    this.predicate = predicate;
    this.inner = inner;
    this.serialSubscription = serialSubscription;
    this.pa = pa;
}

public Subscriber<? super Observable<T>> call(Subscriber<? super T> child) {
    Worker inner = Schedulers.trampoline().createWorker();
    child.add(inner);
    SerialSubscription serialSubscription = new SerialSubscription();
    child.add(serialSubscription);
    ProducerArbiter pa = new ProducerArbiter();
    child.setProducer(pa);
    return new SourceSubscriber(child, this.predicate, inner, serialSubscription, pa);
}

public void onNext(TRight args) {
    synchronized (ResultSink.this.guard) {
        ResultSink resultSink = ResultSink.this;
        int id = resultSink.rightId;
        resultSink.rightId = id + 1;
        ResultSink.this.rightMap.put(Integer.valueOf(id), args);
        int highLeftId = ResultSink.this.leftId;
    }
    ResultSink.this.group.add(new SerialSubscription());
    try {
        Observable<TRightDuration> duration = (Observable) OnSubscribeJoin.this.rightDurationSelector.call(args);
        Subscriber<TRightDuration> d2 = new RightDurationSubscriber(id);
        ResultSink.this.group.add(d2);
        duration.unsafeSubscribe(d2);
        List<TLeft> leftValues = new ArrayList();
        synchronized (ResultSink.this.guard) {
            for (Entry<Integer, TLeft> entry : ResultSink.this.leftMap.entrySet()) {
                if (((Integer) entry.getKey()).intValue() < highLeftId) {
                    leftValues.add(entry.getValue());
                }
            }
        }
        for (TLeft lv : leftValues) {
            ResultSink.this.subscriber.onNext(OnSubscribeJoin.this.resultSelector.call(lv, args));
        }
    } catch (Throwable t) {
        Exceptions.throwOrReport(t, this);
    }
}

private TimeoutSubscriber(SerializedSubscriber<T> serializedSubscriber, TimeoutStub<T> timeoutStub, SerialSubscription serial, Observable<? extends T> other, Worker inner) {
    super(serializedSubscriber);
    this.gate = new Object();
    this.terminated = new AtomicInteger();
    this.actual = new AtomicLong();
    this.serializedSubscriber = serializedSubscriber;
    this.timeoutStub = timeoutStub;
    this.serial = serial;
    this.other = other;
    this.inner = inner;
}

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;
}

SwitchSubscriber(Subscriber<? super T> child) {
    this.serializedChild = new SerializedSubscriber(child);
    this.arbiter = new ProducerArbiter();
    this.ssub = new SerialSubscription();
    child.add(this.ssub);
    child.setProducer(new Producer() {
        public void request(long n) {
            if (n > 0) {
                SwitchSubscriber.this.arbiter.request(n);
            }
        }
    });
}

public Subscriber<? super T> call(Subscriber<? super T> child) {
    SerialSubscription ssub = new SerialSubscription();
    ProducerArbiter arbiter = new ProducerArbiter();
    ParentSubscriber<T> parent = new ParentSubscriber(child, ssub, arbiter, this.alternate);
    ssub.set(parent);
    child.add(ssub);
    child.setProducer(arbiter);
    return parent;
}

public ConcatSubscriber(Subscriber<T> s, SerialSubscription current) {
    super(s);
    this.child = s;
    this.current = current;
    this.arbiter = new ProducerArbiter();
    this.queue = new ConcurrentLinkedQueue();
    add(Subscriptions.create(new Action0() {
        public void call() {
            ConcatSubscriber.this.queue.clear();
        }
    }));
}

public Subscriber<? super Observable<? extends T>> call(Subscriber<? super T> child) {
    SerializedSubscriber<T> s = new SerializedSubscriber(child);
    SerialSubscription current = new SerialSubscription();
    child.add(current);
    ConcatSubscriber<T> cs = new ConcatSubscriber(s, current);
    child.setProducer(new ConcatProducer(cs));
    return cs;
}

public Subscriber<? super T> call(Subscriber<? super T> child) {
    final Worker worker = this.scheduler.createWorker();
    final SerializedSubscriber<T> s = new SerializedSubscriber(child);
    final SerialSubscription ssub = new SerialSubscription();
    s.add(worker);
    s.add(ssub);
    return new Subscriber<T>(child) {
        final Subscriber<?> self = this;
        final DebounceState<T> state = new DebounceState();

        public void onStart() {
            request(Long.MAX_VALUE);
        }

        public void onNext(T t) {
            final int index = this.state.next(t);
            ssub.set(worker.schedule(new Action0() {
                public void call() {
                    AnonymousClass1.this.state.emit(index, s, AnonymousClass1.this.self);
                }
            }, OperatorDebounceWithTime.this.timeout, OperatorDebounceWithTime.this.unit));
        }

        public void onError(Throwable e) {
            s.onError(e);
            unsubscribe();
            this.state.clear();
        }

        public void onCompleted() {
            this.state.emitAndComplete(s, this);
        }
    };
}

public CompletableConcatSubscriber(CompletableSubscriber actual, int prefetch) {
    this.actual = actual;
    this.prefetch = prefetch;
    this.queue = new SpscArrayQueue<Completable>(prefetch);
    this.sr = new SerialSubscription();
    this.inner = new ConcatInnerSubscriber();
    this.wip = new AtomicInteger();
    add(sr);
    request(prefetch);
}

public CompletableConcatSubscriber(CompletableSubscriber actual, int prefetch) {
    this.actual = actual;
    this.prefetch = prefetch;
    this.queue = new SpscArrayQueue<Completable>(prefetch);
    this.sr = new SerialSubscription();
    this.inner = new ConcatInnerSubscriber();
    this.wip = new AtomicInteger();
    add(sr);
    request(prefetch);
}

@Override
public Subscriber<? super Observable<T>> call(final Subscriber<? super T> child) {
    final Scheduler.Worker inner = Schedulers.trampoline().createWorker();
    child.add(inner);

    final SerialSubscription serialSubscription = new SerialSubscription();
    // add serialSubscription so it gets unsubscribed if child is unsubscribed
    child.add(serialSubscription);

    return new SourceSubscriber<T>(child, predicate, inner, serialSubscription);
}

public SourceSubscriber(Subscriber<? super T> child, final BiFunction<Integer, Throwable, Boolean> predicate, Scheduler.Worker inner, 
        SerialSubscription serialSubscription) {
    this.child = child;
    this.predicate = predicate;
    this.inner = inner;
    this.serialSubscription = serialSubscription;
}

@Override
public void onNext(TRight args) {
    int id; 
    int highLeftId;
    synchronized (guard) {
        id = rightId++;
        rightMap.put(id, args);
        highLeftId = leftId;
    }
    SerialSubscription md = new SerialSubscription();
    group.add(md);

    Observable<TRightDuration> duration;
    try {
        duration = rightDurationSelector.call(args);

        Subscriber<TRightDuration> d2 = new RightDurationSubscriber(id);
        group.add(d2);

        duration.unsafeSubscribe(d2);


        List<TLeft> leftValues = new ArrayList<TLeft>();
        synchronized (guard) {
            for (Map.Entry<Integer, TLeft> entry : leftMap.entrySet()) {
                if (entry.getKey() < highLeftId) {
                    leftValues.add(entry.getValue());
                }
            }
        }

        for (TLeft lv : leftValues) {
            R result = resultSelector.call(lv, args);
            subscriber.onNext(result);
        }

    } catch (Throwable t) {
        onError(t);
    }
}

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);
            }
        }
    });
}

@Override
public Subscriber<? super T> call(Subscriber<? super T> child) {
    final SerialSubscription ssub = new SerialSubscription();
    final SwitchIfEmptySubscriber parent = new SwitchIfEmptySubscriber(child, ssub);
    ssub.set(parent);
    child.add(ssub);
    return parent;
}

@Override
public Subscriber<? super Observable<? extends T>> call(final Subscriber<? super T> child) {
    final SerializedSubscriber<T> s = new SerializedSubscriber<T>(child);
    final SerialSubscription current = new SerialSubscription();
    child.add(current);
    ConcatSubscriber<T> cs = new ConcatSubscriber<T>(s, current);
    ConcatProducer<T> cp = new ConcatProducer<T>(cs);
    child.setProducer(cp);
    return cs;
}

public ConcatSubscriber(Subscriber<T> s, SerialSubscription current) {
    super(s);
    this.child = s;
    this.current = current;
    this.queue = new ConcurrentLinkedQueue<Object>();
    add(Subscriptions.create(new Action0() {
        @Override
        public void call() {
            queue.clear();
        }
    }));
}

@Override
public Subscriber<? super T> call(final Subscriber<? super T> child) {
    final Worker worker = scheduler.createWorker();
    final SerializedSubscriber<T> s = new SerializedSubscriber<T>(child);
    final SerialSubscription ssub = new SerialSubscription();

    s.add(worker);
    s.add(ssub);

    return new Subscriber<T>(child) {
        final DebounceState<T> state = new DebounceState<T>();
        final Subscriber<?> self = this;
        @Override
        public void onNext(final T t) {

            final int index = state.next(t);
            ssub.set(worker.schedule(new Action0() {
                @Override
                public void call() {
                    state.emit(index, s, self);
                }
            }, timeout, unit));
        }

        @Override
        public void onError(Throwable e) {
            s.onError(e);
            unsubscribe();
            state.clear();
        }

        @Override
        public void onComplete() {
            state.emitAndComplete(s, this);
        }
    };
}

public ChainSubscriber(SerialSubscription subscription, ExecutionContext context, I request, O response,
                       Subscriber<? super Void> child) {
    this.subscription = subscription;
    this.context = context;
    this.request = request;
    this.response = response;
    this.child = child;
}

@Override
public Subscriber<? super Observable<T>> call(final Subscriber<? super T> child) {
    final Scheduler.Worker inner = Schedulers.trampoline().createWorker();
    child.add(inner);

    final SerialSubscription serialSubscription = new SerialSubscription();
    // add serialSubscription so it gets unsubscribed if child is unsubscribed
    child.add(serialSubscription);

    return new SourceSubscriber<T>(child, predicate, inner, serialSubscription);
}

public SourceSubscriber(Subscriber<? super T> child, final Func2<Integer, Throwable, Boolean> predicate, Scheduler.Worker inner, 
        SerialSubscription serialSubscription) {
    this.child = child;
    this.predicate = predicate;
    this.inner = inner;
    this.serialSubscription = serialSubscription;
}

@Override
public void onNext(TRight args) {
    int id; 
    int highLeftId;
    synchronized (guard) {
        id = rightId++;
        rightMap.put(id, args);
        highLeftId = leftId;
    }
    SerialSubscription md = new SerialSubscription();
    group.add(md);

    Observable<TRightDuration> duration;
    try {
        duration = rightDurationSelector.call(args);

        Subscriber<TRightDuration> d2 = new RightDurationSubscriber(id);
        group.add(d2);

        duration.unsafeSubscribe(d2);


        List<TLeft> leftValues = new ArrayList<TLeft>();
        synchronized (guard) {
            for (Map.Entry<Integer, TLeft> entry : leftMap.entrySet()) {
                if (entry.getKey() < highLeftId) {
                    leftValues.add(entry.getValue());
                }
            }
        }

        for (TLeft lv : leftValues) {
            R result = resultSelector.call(lv, args);
            subscriber.onNext(result);
        }

    } catch (Throwable t) {
        onError(t);
    }
}

@Override
public Subscriber<? super T> call(Subscriber<? super T> subscriber) {
    Scheduler.Worker inner = scheduler.createWorker();
    subscriber.add(inner);
    final SerialSubscription serial = new SerialSubscription();
    subscriber.add(serial);
    // Use SynchronizedSubscriber for safe memory access
    // as the subscriber will be accessed in the current thread or the
    // scheduler or other Observables.
    final SerializedSubscriber<T> synchronizedSubscriber = new SerializedSubscriber<T>(subscriber);

    TimeoutSubscriber<T> timeoutSubscriber = new TimeoutSubscriber<T>(synchronizedSubscriber, timeoutStub, serial, other, inner);
    serial.set(firstTimeoutStub.call(timeoutSubscriber, 0L, inner));
    return timeoutSubscriber;
}

private TimeoutSubscriber(
        SerializedSubscriber<T> serializedSubscriber,
        TimeoutStub<T> timeoutStub, SerialSubscription serial,
        Observable<? extends T> other,
        Scheduler.Worker inner) {
    super(serializedSubscriber);
    this.serializedSubscriber = serializedSubscriber;
    this.timeoutStub = timeoutStub;
    this.serial = serial;
    this.other = other;
    this.inner = inner;
}

@Override
public Subscriber<? super Observable<? extends T>> call(final Subscriber<? super T> child) {
    final SerializedSubscriber<T> s = new SerializedSubscriber<T>(child);
    final SerialSubscription current = new SerialSubscription();
    child.add(current);
    ConcatSubscriber<T> cs = new ConcatSubscriber<T>(s, current);
    ConcatProducer<T> cp = new ConcatProducer<T>(cs);
    child.setProducer(cp);
    return cs;
}

public ConcatSubscriber(Subscriber<T> s, SerialSubscription current) {
    super(s);
    this.child = s;
    this.current = current;
    this.queue = new ConcurrentLinkedQueue<Object>();
    add(Subscriptions.create(new Action0() {
        @Override
        public void call() {
            queue.clear();
        }
    }));
}

@Override
public Subscriber<? super T> call(final Subscriber<? super T> child) {
    final Worker worker = scheduler.createWorker();
    final SerializedSubscriber<T> s = new SerializedSubscriber<T>(child);
    final SerialSubscription ssub = new SerialSubscription();

    s.add(worker);
    s.add(ssub);

    return new Subscriber<T>(child) {
        final DebounceState<T> state = new DebounceState<T>();
        final Subscriber<?> self = this;
        @Override
        public void onNext(final T t) {

            final int index = state.next(t);
            ssub.set(worker.schedule(new Action0() {
                @Override
                public void call() {
                    state.emit(index, s, self);
                }
            }, timeout, unit));
        }

        @Override
        public void onError(Throwable e) {
            s.onError(e);
            unsubscribe();
            state.clear();
        }

        @Override
        public void onCompleted() {
            state.emitAndComplete(s, this);
        }
    };
}