OperatorDebounceWithTime.java 文件源码

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