AtomicRateLimiter implements RateLimiter { @Override public boolean acquirePermission(final int permits) { long timeoutInNanos = state.get().config.getTimeoutDuration().toNanos(); State modifiedState = updateStateWithBackOff(permits, timeoutInNanos); boolean result = waitForPermissionIfNecessary(timeoutInNanos, modifiedState.nanosToWait); publishRateLimiterEvent(result, permits); return result; } AtomicRateLimiter(String name, RateLimiterConfig rateLimiterConfig);  AtomicRateLimiter(String name, RateLimiterConfig rateLimiterConfig,
                             Map<String, String> tags); @Override void changeTimeoutDuration(final Duration timeoutDuration); @Override void changeLimitForPeriod(final int limitForPeriod); @Override boolean acquirePermission(final int permits); @Override long reservePermission(final int permits); @Override String getName(); @Override RateLimiterConfig getRateLimiterConfig(); @Override Map<String, String> getTags(); @Override Metrics getMetrics(); @Override EventPublisher getEventPublisher(); @Override String toString(); AtomicRateLimiterMetrics getDetailedMetrics();  }
@Test public void waitingThreadIsInterrupted() throws Exception { setup(Duration.ofNanos(CYCLE_IN_NANOS)); setTimeOnNanos(CYCLE_IN_NANOS); boolean permission = rateLimiter.acquirePermission(); then(permission).isTrue(); then(metrics.getAvailablePermissions()).isEqualTo(0); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS); then(metrics.getNumberOfWaitingThreads()).isEqualTo(0); AtomicReference<Boolean> reservedPermission = new AtomicReference<>(null); AtomicBoolean wasInterrupted = new AtomicBoolean(false); Thread caller = new Thread( () -> { reservedPermission.set(rateLimiter.acquirePermission()); wasInterrupted.set(Thread.currentThread().isInterrupted()); } ); caller.setDaemon(true); caller.start(); awaitImpatiently() .atMost(5, SECONDS) .until(caller::getState, equalTo(Thread.State.TIMED_WAITING)); then(metrics.getAvailablePermissions()).isEqualTo(-1); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS * 2); then(metrics.getNumberOfWaitingThreads()).isEqualTo(1); caller.interrupt(); awaitImpatiently() .atMost(5, SECONDS) .until(reservedPermission::get, equalTo(false)); then(wasInterrupted.get()).isTrue(); then(metrics.getAvailablePermissions()).isEqualTo(-1); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS * 2); then(metrics.getNumberOfWaitingThreads()).isEqualTo(0); }
@Test public void acquireAndRefreshWithEventPublishing() throws Exception { setup(Duration.ZERO); setTimeOnNanos(CYCLE_IN_NANOS); boolean permission = rateLimiter.acquirePermission(); then(permission).isTrue(); then(metrics.getAvailablePermissions()).isEqualTo(0); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS); boolean secondPermission = rateLimiter.acquirePermission(); then(secondPermission).isFalse(); then(metrics.getAvailablePermissions()).isEqualTo(0); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS); setTimeOnNanos(CYCLE_IN_NANOS * 2); boolean thirdPermission = rateLimiter.acquirePermission(); then(thirdPermission).isTrue(); then(metrics.getAvailablePermissions()).isEqualTo(0); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS); boolean fourthPermission = rateLimiter.acquirePermission(); then(fourthPermission).isFalse(); then(metrics.getAvailablePermissions()).isEqualTo(0); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS); }
@Test public void reserveAndRefresh() throws Exception { setup(Duration.ofNanos(CYCLE_IN_NANOS)); setTimeOnNanos(CYCLE_IN_NANOS); boolean permission = rateLimiter.acquirePermission(); then(permission).isTrue(); then(metrics.getAvailablePermissions()).isEqualTo(0); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS); AtomicReference<Boolean> reservedPermission = new AtomicReference<>(null); Thread caller = new Thread( () -> reservedPermission.set(rateLimiter.acquirePermission())); caller.setDaemon(true); caller.start(); awaitImpatiently() .atMost(5, SECONDS) .until(caller::getState, equalTo(Thread.State.TIMED_WAITING)); then(metrics.getAvailablePermissions()).isEqualTo(-1); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS + CYCLE_IN_NANOS); then(metrics.getNumberOfWaitingThreads()).isEqualTo(1); setTimeOnNanos(CYCLE_IN_NANOS * 2 + 10); awaitImpatiently() .atMost(5, SECONDS) .until(reservedPermission::get, equalTo(true)); then(metrics.getAvailablePermissions()).isEqualTo(0); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS - 10); then(metrics.getNumberOfWaitingThreads()).isEqualTo(0); }
@Test public void reserveFewThenSkipCyclesBeforeRefresh() throws Exception { setup(Duration.ofNanos(CYCLE_IN_NANOS)); setTimeOnNanos(CYCLE_IN_NANOS); boolean permission = rateLimiter.acquirePermission(); then(permission).isTrue(); then(metrics.getAvailablePermissions()).isEqualTo(0); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS); then(metrics.getNumberOfWaitingThreads()).isEqualTo(0); AtomicReference<Boolean> firstReservedPermission = new AtomicReference<>(null); Thread firstCaller = new Thread( () -> firstReservedPermission.set(rateLimiter.acquirePermission())); firstCaller.setDaemon(true); firstCaller.start(); awaitImpatiently() .atMost(5, SECONDS) .until(firstCaller::getState, equalTo(Thread.State.TIMED_WAITING)); then(metrics.getAvailablePermissions()).isEqualTo(-1); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS * 2); then(metrics.getNumberOfWaitingThreads()).isEqualTo(1); AtomicReference<Boolean> secondReservedPermission = new AtomicReference<>(null); Thread secondCaller = new Thread( () -> secondReservedPermission.set(rateLimiter.acquirePermission())); secondCaller.setDaemon(true); secondCaller.start(); awaitImpatiently() .atMost(5, SECONDS) .until(secondCaller::getState, equalTo(Thread.State.TIMED_WAITING)); then(metrics.getAvailablePermissions()).isEqualTo(-1); then(metrics.getNanosToWait()).isEqualTo(CYCLE_IN_NANOS * 2); then(metrics.getNumberOfWaitingThreads()).isEqualTo(2); setTimeOnNanos(CYCLE_IN_NANOS * 6 + 10); awaitImpatiently() .atMost(5, SECONDS) .until(firstReservedPermission::get, equalTo(true)); awaitImpatiently() .atMost(5, SECONDS) .until(secondReservedPermission::get, equalTo(false)); then(metrics.getAvailablePermissions()).isEqualTo(1); then(metrics.getNanosToWait()).isEqualTo(0L); then(metrics.getNumberOfWaitingThreads()).isEqualTo(0); }