ThrottledCall extends Call.Base<Void> { @Override public Call<Void> clone() { return new ThrottledCall(delegate.clone(), executor, limiter, limiterMetrics, isOverCapacity); } ThrottledCall(Call<Void> delegate, Executor executor, Limiter<Void> limiter,
    LimiterMetrics limiterMetrics, Predicate<Throwable> isOverCapacity); @Override Call<Void> clone(); @Override String toString();  }
@Test public void execute_isThrottled() throws Exception { int queueSize = 1; int totalTasks = numThreads + queueSize; limit.setLimit(totalTasks); Semaphore startLock = new Semaphore(numThreads); Semaphore waitLock = new Semaphore(totalTasks); Semaphore failLock = new Semaphore(1); ThrottledCall throttled = throttle(new LockedCall(startLock, waitLock)); startLock.drainPermits(); waitLock.drainPermits(); failLock.drainPermits(); ExecutorService backgroundPool = Executors.newCachedThreadPool(); for (int i = 0; i < totalTasks; i++) { backgroundPool.submit(() -> throttled.clone().execute()); } try { startLock.acquire(numThreads); Future<?> future = backgroundPool.submit(() -> { try { throttled.execute(); } catch (IOException e) { throw new UncheckedIOException(e); } finally { failLock.release(); } }); failLock.acquire(); future.get(); failBecauseExceptionWasNotThrown(ExecutionException.class); } catch (ExecutionException t) { assertThat(t) .isInstanceOf(ExecutionException.class) .hasCauseInstanceOf(RejectedExecutionException.class); } finally { waitLock.release(totalTasks); startLock.release(totalTasks); backgroundPool.shutdownNow(); } }
@Test public void enqueue_isThrottled() throws Exception { int queueSize = 1; int totalTasks = numThreads + queueSize; limit.setLimit(totalTasks); Semaphore startLock = new Semaphore(numThreads); Semaphore waitLock = new Semaphore(totalTasks); ThrottledCall throttle = throttle(new LockedCall(startLock, waitLock)); startLock.drainPermits(); waitLock.drainPermits(); Callback<Void> callback = mock(Callback.class); for (int i = 0; i < totalTasks; i++) { throttle.clone().enqueue(callback); } startLock.acquire(numThreads); assertThatThrownBy(() -> throttle.clone().enqueue(callback)) .isEqualTo(STORAGE_THROTTLE_MAX_CONCURRENCY); }