OnSubscribeExponentialDelay.java 文件源码

public Subscription schedulePeriodically(
    final Scheduler.Worker worker,
    final Action0 action,
    long initialDelay,
    Delay periodDelay,
    final TimeUnit unit) {
  final Delay period = periodDelay;
  final long firstNowNanos = TimeUnit.MILLISECONDS.toNanos(worker.now());
  final long firstStartInNanos = firstNowNanos + unit.toNanos(initialDelay);

  final MultipleAssignmentSubscription mas = new MultipleAssignmentSubscription();
  final Action0 recursiveAction =
      new Action0() {
        long count;
        long lastNowNanos = firstNowNanos;
        long startInNanos = firstStartInNanos;

        @Override
        public void call() {
          if (!mas.isUnsubscribed()) {
            action.call();
            long periodInNanos = unit.toNanos(period.calculate(count));
            long nextTick;

            long nowNanos = TimeUnit.MILLISECONDS.toNanos(worker.now());
            // If the clock moved in a direction quite a bit, rebase the repetition period
            if (nowNanos + CLOCK_DRIFT_TOLERANCE_NANOS < lastNowNanos
                || nowNanos >= lastNowNanos + periodInNanos + CLOCK_DRIFT_TOLERANCE_NANOS) {
              nextTick = nowNanos + periodInNanos;
              /*
               * Shift the start point back by the drift as if the whole thing
               * started count periods ago.
               */
              startInNanos = nextTick - (periodInNanos * (++count));
            } else {
              nextTick = startInNanos + (++count * periodInNanos);
            }
            lastNowNanos = nowNanos;

            long delay = nextTick - nowNanos;
            mas.set(worker.schedule(this, delay, TimeUnit.NANOSECONDS));
          }
        }
      };
  MultipleAssignmentSubscription s = new MultipleAssignmentSubscription();
  // Should call `mas.set` before `schedule`, or the new Subscription may replace the old one.
  mas.set(s);
  s.set(worker.schedule(recursiveAction, initialDelay, unit));
  return mas;
}