@@ -0,0 +1,3475 @@
+package forklift.concurrent;%0A%0Aimport static org.mockito.Mockito.any;%0Aimport static org.mockito.Mockito.times;%0A%0Aimport org.junit.After;%0Aimport org.junit.Assert;%0Aimport org.junit.Before;%0Aimport org.junit.Test;%0Aimport org.mockito.Mockito;%0A%0Aimport java.util.concurrent.ArrayBlockingQueue;%0Aimport java.util.concurrent.BlockingQueue;%0Aimport java.util.concurrent.Executors;%0Aimport java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.RejectedExecutionHandler;%0Aimport java.util.concurrent.ThreadPoolExecutor;%0Aimport java.util.concurrent.TimeUnit;%0Aimport java.util.concurrent.atomic.AtomicBoolean;%0A%0Apublic class BlockingRejectedExecutionHandlerTests %7B%0A    private ThreadPoolExecutor pool;%0A    private BlockingQueue%3CRunnable%3E queue;%0A    private RejectedExecutionHandler handler;%0A%0A    @Before%0A    public void setup() %7B%0A        this.queue = new ArrayBlockingQueue%3C%3E(1);%0A        this.pool = new ThreadPoolExecutor(1, 1, 10, TimeUnit.MILLISECONDS, this.queue);%0A        this.handler = Mockito.spy(new BlockingRejectedExecutionHandler());%0A        this.pool.setRejectedExecutionHandler(handler);%0A    %7D%0A%0A    @After%0A    public void shutdown() throws Exception %7B%0A        pool.shutdownNow();%0A        pool.awaitTermination(100, TimeUnit.MILLISECONDS);%0A    %7D%0A%0A    @Test%0A    public void testRejectedExecutionTriggered() %7B%0A        final AtomicBoolean blockCondition = new AtomicBoolean(false);%0A        final AtomicBoolean finishCondition = new AtomicBoolean(false);%0A%0A        // fill up the queue and executing threads - nothing should be rejected yet%0A        startBlockedTask(blockCondition, pool);%0A        startBlockedTask(blockCondition, pool);%0A        Mockito.verify(handler, times(0)).rejectedExecution(any(), any());%0A%0A        doAsync(() -%3E %7B%0A            pool.execute(() -%3E %7B%0A                synchronized (finishCondition) %7B%0A                    finishCondition.set(true);%0A                    finishCondition.notifyAll();%0A                %7D%0A            %7D);%0A        %7D);%0A%0A        // make sure that the third task on the pool doesn't get executed yet, but the rejected execution handler has started%0A        synchronized (finishCondition) %7B%0A            try %7B%0A                finishCondition.wait(100);%0A            %7D catch (InterruptedException ignored) %7B%7D%0A        %7D%0A        Assert.assertFalse(finishCondition.get());%0A        Mockito.verify(handler, times(1)).rejectedExecution(any(), any());%0A%0A        // unblock the queued threads%0A        synchronized (blockCondition) %7B%0A            blockCondition.set(true);%0A            blockCondition.notifyAll();%0A        %7D%0A%0A        // check that the queued task eventually gets ran%0A        synchronized (finishCondition) %7B%0A            try %7B%0A                finishCondition.wait(100);%0A            %7D catch (InterruptedException ignored) %7B%7D%0A        %7D%0A        Assert.assertTrue(finishCondition.get());%0A        Mockito.verify(handler, times(1)).rejectedExecution(any(), any());%0A    %7D%0A%0A    private void startBlockedTask(AtomicBoolean condition, ExecutorService executor) %7B%0A        executor.submit(() -%3E %7B%0A            while (!condition.get()) %7B%0A                synchronized (condition) %7B%0A                    try %7B%0A                        condition.wait();%0A                    %7D catch (InterruptedException ignored) %7B%7D%0A                %7D%0A            %7D%0A        %7D);%0A    %7D%0A%0A    private void doAsync(Runnable action) %7B%0A        final ExecutorService service =  Executors.newSingleThreadExecutor();%0A        service.execute(action);%0A        service.shutdown();%0A    %7D%0A%7D%0A

Test that the blocking execution handler actually blocks