ThreadPoolService implements Precipice<TimeoutableResult, Rejected> { public <T> PrecipiceFuture<TimeoutableResult, T> submit(Callable<T> callable) { PrecipicePromise<TimeoutableResult, T> promise = Asynchronous.acquirePermitsAndPromise(guardRail, 1L); internalComplete(callable, promise); return promise.future(); } ThreadPoolService(int poolSize, int queueSize, GuardRail<TimeoutableResult, Rejected> guardRail);  ThreadPoolService(ExecutorService executorService, GuardRail<TimeoutableResult, Rejected> guardRail); @Override GuardRail<TimeoutableResult, Rejected> guardRail(); PrecipiceFuture<TimeoutableResult, T> submit(Callable<T> callable); PrecipiceFuture<TimeoutableResult, T> submit(Callable<T> callable, long millisTimeout); void complete(Callable<T> callable, PrecipicePromise<TimeoutableResult, T> promise); void complete(Callable<T> callable, PrecipicePromise<TimeoutableResult, T> promise, long millisTimeout); ExecutorService getExecutor(); DelayQueueTimeoutService getTimeoutService(); void shutdown();  }
@Test public void guardRailsClockIsUsedForStartTime() { long startNanos = 100L; Clock clock = mock(Clock.class); when(guardRail.getClock()).thenReturn(clock); when(clock.nanoTime()).thenReturn(startNanos); service.submit(TestCallable.success(), 100); verify(guardRail).acquirePermits(1L, startNanos); }
@Test public void simulationTest() { GuardRailBuilder<TimeoutableResult, SimulationRejected> builder = new GuardRailBuilder<TimeoutableResult, SimulationRejected>() .name("Simulation") .resultCounts(new TotalCounts<TimeoutableResult>(TimeoutableResult.class)) .rejectedCounts(new TotalCounts<>(SimulationRejected.class)) .resultLatency(new NoOpLatency<>(TimeoutableResult.class)); GuardRail<TimeoutableResult, SimulationRejected> guardRail = builder.build(); final ThreadPoolService<SimulationRejected> callService = new ThreadPoolService<>(5, 10, guardRail); Map<TimeoutableResult, Callable<Long>> resultToCallable = new HashMap<>(); resultToCallable.put(TimeoutableResult.SUCCESS, new Callable<Long>() { @Override public Long call() throws InterruptedException { try { PrecipiceFuture<TimeoutableResult, String> f = callService.submit(TestCallable.success()); f.await(); } catch (RejectedException e) { } return 1L; } }); resultToCallable.put(TimeoutableResult.ERROR, new Callable<Long>() { @Override public Long call() throws InterruptedException { try { PrecipiceFuture<TimeoutableResult, String> f = callService.submit(TestCallable.erred(new IOException())); f.await(); } catch (RejectedException e) { } return 1L; } }); resultToCallable.put(TimeoutableResult.TIMEOUT, new Callable<Long>() { @Override public Long call() throws InterruptedException { try { CountDownLatch latch = new CountDownLatch(1); PrecipiceFuture<TimeoutableResult, String> f = callService.submit(TestCallable.blocked(latch), 1L); f.await(); latch.countDown(); } catch (RejectedException e) { } return 1L; } }); Simulation<TimeoutableResult> simulation = new Simulation<>(guardRail); simulation.run(resultToCallable); }
@Test public void exceptionThrownIfControllerRejects() throws Exception { try { when(guardRail.acquirePermits(eq(1L), anyLong())).thenThrow(new RejectedException(Rejected .MAX_CONCURRENCY_LEVEL_EXCEEDED)); service.submit(TestCallable.success(), Long.MAX_VALUE); fail(); } catch (RejectedException e) { assertEquals(Rejected.MAX_CONCURRENCY_LEVEL_EXCEEDED, e.reason); } try { when(guardRail.acquirePermits(eq(1L), anyLong())).thenThrow(new RejectedException(Rejected.CIRCUIT_OPEN)); service.submit(TestCallable.success(), Long.MAX_VALUE); fail(); } catch (RejectedException e) { assertEquals(Rejected.CIRCUIT_OPEN, e.reason); } }
@Test public void callableIsSubmittedAndRan() throws Exception { when(guardRail.acquirePermits(eq(1L), anyLong())).thenReturn(null); PrecipiceFuture<TimeoutableResult, String> f = service.submit(TestCallable.success(), DelayQueueTimeoutService.MAX_TIMEOUT_MILLIS); assertEquals("Success", f.get()); assertEquals(TimeoutableResult.SUCCESS, f.getResult()); }
@Test public void submittedCallableWillTimeout() throws Exception { when(guardRail.acquirePermits(eq(1L), anyLong())).thenReturn(null); CountDownLatch latch = new CountDownLatch(1); PrecipiceFuture<TimeoutableResult, String> future = service.submit(TestCallable.blocked(latch), 1); try { future.get(); fail("Should have thrown ExecutionException from PrecipiceTimeoutException"); } catch (ExecutionException e) { assertTrue(e.getCause() instanceof PrecipiceTimeoutException); } assertEquals(TimeoutableResult.TIMEOUT, future.getResult()); assertTrue(future.getError() instanceof PrecipiceTimeoutException); latch.countDown(); }
@Test public void erredCallableWillReturnException() { when(guardRail.acquirePermits(eq(1L), anyLong())).thenReturn(null); RuntimeException exception = new RuntimeException(); PrecipiceFuture<TimeoutableResult, String> future = service.submit(TestCallable.erred(exception), 100); try { future.get(); fail(); } catch (InterruptedException e) { fail(); } catch (ExecutionException e) { assertEquals(exception, e.getCause()); } assertEquals(exception, future.getError()); assertNull(future.getValue()); assertEquals(TimeoutableResult.ERROR, future.getResult()); }