BlockingExecutor extends AbstractExecutorService implements ExecutorService { @Override public synchronized void execute(Runnable command) { try { semaphore.acquire(); } catch (InterruptedException e) { throw new RuntimeException("Interrupted while acquiring permit.", e); } AtomicBoolean released = new AtomicBoolean(false); Runnable release = () -> { if (!released.getAndSet(true)) { semaphore.release(); } }; try { delegate.execute(() -> { try { command.run(); } finally { release.run(); } }); } catch (Exception e) { release.run(); throw e; } } BlockingExecutor(int poolSize, int queueSize); @Override synchronized void execute(Runnable command); @Override void shutdown(); @Override List<Runnable> shutdownNow(); @Override boolean awaitTermination(long timeout, TimeUnit unit); @Override boolean isShutdown(); @Override boolean isTerminated();  }
@Test public void executeHappyPath() throws InterruptedException { underTest = new BlockingExecutor(delegate, semaphore); underTest.execute(runnable); InOrder inOrder = inOrder(delegate, semaphore); inOrder.verify(semaphore).acquire(); inOrder.verify(delegate).execute(captor.capture()); inOrder.verify(semaphore, never()).release(); captor.getValue().run(); InOrder inOrder2 = inOrder(runnable, semaphore); inOrder2.verify(runnable).run(); inOrder2.verify(semaphore).release(); }
@Test public void delegateException() throws InterruptedException { doThrow(RuntimeException.class).when(delegate).execute(any(Runnable.class)); underTest = new BlockingExecutor(delegate, semaphore); try { underTest.execute(runnable); fail(); } catch (RuntimeException e) { InOrder inOrder = inOrder(delegate, semaphore); inOrder.verify(semaphore).acquire(); inOrder.verify(delegate).execute(any(Runnable.class)); inOrder.verify(semaphore).release(); } }
@Test public void runnableException() throws InterruptedException { doThrow(RuntimeException.class).when(runnable).run(); underTest = new BlockingExecutor(delegate, semaphore); underTest.execute(runnable); InOrder inOrder = inOrder(delegate, semaphore); inOrder.verify(semaphore).acquire(); inOrder.verify(delegate).execute(captor.capture()); inOrder.verify(semaphore, never()).release(); try { captor.getValue().run(); fail(); } catch (RuntimeException e) { InOrder inOrder2 = inOrder(runnable, semaphore); inOrder2.verify(runnable).run(); inOrder2.verify(semaphore).release(); } }
@Test(expected = RuntimeException.class) public void semaphoreInterruptedException() throws InterruptedException { underTest = new BlockingExecutor(delegate, semaphore); doThrow(InterruptedException.class).when(semaphore).acquire(); underTest.execute(runnable); }