@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(); } }
@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 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 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); }
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(); }
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();  }