LimitingExecutor implements Executor { @Override public void execute(final Runnable command) { try { execute((C) Callables.from(command)); } catch (Exception ex) { throw new UncheckedExecutionException(ex); } } LimitingExecutor(final PermitSupplier permitSupplier);  LimitingExecutor(final Semaphore semaphore);  LimitingExecutor(final RejectedExecutionHandler<T, C> rejectHandler, final Semaphore semaphore); @Override void execute(final Runnable command); T execute(final C callable); Callable<T> toLimitedCallable(final C callable); RejectedExecutionHandler getRejectHandler(); Semaphore getSemaphore(); @Override String toString();  }
@Test(expected = RejectedExecutionException.class) public void testRateLimit() throws Exception { try (RateLimiter limiter = new RateLimiter(10, Duration.ofSeconds(1), 10)) { LimitingExecutor<?, Callable<?>> executor = new LimitingExecutor<>(limiter); Assert.assertEquals(10, limiter.getPermitsPerReplenishInterval()); Assert.assertEquals(1000000000, limiter.getPermitReplenishIntervalNanos()); for (int i = 0; i < 10; i++) { final int val = i; executor.execute(() -> { LOG.debug("executed nr {}", val); return null; }); } Assert.fail(); } }
@Test public void testRateLimit2() throws Exception { LogAssert expect = TestLoggers.sys().expect(LimitingExecutorTest.class.getName(), Level.DEBUG, 10, LogMatchers.hasFormat("executed nr {}")); try (RateLimiter limiter = new RateLimiter(10, Duration.ofSeconds(1), 10)) { long replenishInterval = limiter.getPermitReplenishIntervalNanos(); LOG.debug("replenish interval = {} ", replenishInterval); LimitingExecutor.RejectedExecutionHandler rejectedExecutionHandler = new LimitingExecutor.RejectedExecutionHandler() { @Override @SuppressFBWarnings("MDM_THREAD_YIELD") public Object reject(final LimitingExecutor executor, final Callable callable) throws Exception { long nanoTime = TimeSource.nanoTime(); long lastReplenishmentNanos = limiter.getLastReplenishmentNanos(); long waitNs = replenishInterval - (nanoTime - lastReplenishmentNanos); if (waitNs >= 0) { TimeUnit.NANOSECONDS.sleep(waitNs); } else { LOG.debug("negative wait time {}, current = {}, lastUpdated = {}, permits = {} ", waitNs, nanoTime, lastReplenishmentNanos, limiter.getNrPermits()); } return executor.execute(callable); } }; LimitingExecutor<?, Callable<?>> executor = new LimitingExecutor<>(rejectedExecutionHandler, limiter.toSemaphore()); for (int i = 0; i < 10; i++) { final int val = i; executor.execute(() -> { LOG.debug("executed nr {}", val); return null; }); } } expect.assertObservation(); }