OperatorDebounceWithTime.java 文件源码

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