ScheduledExecutorServiceMock implements ScheduledExecutorService { @Override public <T> List<Future<T>> invokeAll(final Collection<? extends Callable<T>> tasks, final long timeout, final TimeUnit unit) throws InterruptedException { return doInvokeAll(tasks, Long.valueOf(timeout), unit); } ScheduledExecutorServiceMock();  ScheduledExecutorServiceMock(final ISystemTimeProvider systemTimeProvider); List<ScheduledFutureMock<?>> getEvents(); List<ScheduledFutureMock<?>> executeEvents(); ScheduledFutureMock<?> executeNextEvent(); ScheduledFutureMock<?> offerNextEventForExecution(); @Override void execute(final Runnable runnable); @Override ScheduledFutureMock<Void> submit(final Runnable runnable); @Override ScheduledFutureMock<T> submit(final Runnable runnable, final T result); @Override ScheduledFutureMock<T> submit(final Callable<T> callable); @Override ScheduledFutureMock<Void> schedule(final Runnable runnable, final long delay, final TimeUnit unit); @Override ScheduledFutureMock<T> schedule(final Callable<T> callable, final long delay, final TimeUnit unit); @Override ScheduledFutureMock<Void> scheduleAtFixedRate(
        final Runnable runnable,
        final long initialDelay,
        final long period,
        final TimeUnit unit); @Override ScheduledFutureMock<Void> scheduleWithFixedDelay(
        final Runnable runnable,
        final long initialDelay,
        final long delay,
        final TimeUnit unit); @Override T invokeAny(final Collection<? extends Callable<T>> tasks); @Override T invokeAny(final Collection<? extends Callable<T>> tasks, final long timeout, final TimeUnit unit); @Override List<Future<T>> invokeAll(final Collection<? extends Callable<T>> tasks, final long timeout, final TimeUnit unit); @Override List<Future<T>> invokeAll(final Collection<? extends Callable<T>> tasks); @Override void shutdown(); @Override List<Runnable> shutdownNow(); @Override boolean isShutdown(); @Override boolean isTerminated(); @Override boolean awaitTermination(final long timeout, final TimeUnit unit);  }
@Test public void testInvokeAll() throws InterruptedException, ExecutionException { final ScheduledExecutorServiceMock executor = new ScheduledExecutorServiceMock(); final int taskCount = 500; final int dispatcherCount = 50; final ArrayList<Callable<Integer>> tasks = new ArrayList<Callable<Integer>>(taskCount); for (int i = 0; i < taskCount; i++) { final Integer result = Integer.valueOf(i); tasks.add(new Callable<Integer>() { @Override public Integer call() throws Exception { return result; } }); } final CountDownLatch dispatcherSync = new CountDownLatch(1); final ArrayList<MessageDispatcher> messageDispatchers = new ArrayList<MessageDispatcher>(); final ArrayList<Thread> dispatcherThreads = new ArrayList<Thread>(); for (int i = 0; i < dispatcherCount; i++) { final MessageDispatcher messageDispatcher = new MessageDispatcher(executor, dispatcherSync); messageDispatchers.add(messageDispatcher); final Thread dispatcherThread = new Thread(messageDispatcher, "Dispatcher thread" + i); dispatcherThreads.add(dispatcherThread); dispatcherThread.start(); } dispatcherSync.countDown(); try { final List<Future<Integer>> invocationResults = executor.invokeAll(tasks); Assert.assertEquals(taskCount, invocationResults.size()); for (int i = 0; i < taskCount; i++) { final Future<Integer> invocationResult = invocationResults.get(i); Assert.assertTrue(invocationResult.isDone()); Assert.assertFalse(invocationResult.isCancelled()); Assert.assertEquals(Integer.valueOf(i), invocationResult.get()); } } finally { executor.shutdown(); Assert.assertTrue(executor.awaitTermination(TIMEOUT, TimeUnit.MILLISECONDS)); Assert.assertEquals(0, executor.getEvents().size()); for (final Thread dispatcherThread : dispatcherThreads) { dispatcherThread.join(TIMEOUT); Assert.assertFalse(dispatcherThread.isAlive()); } } }