@Test public void returnsFuturesSetWithTheExceptionsThrownBySubmittedTasks() throws Exception { KeySerializingExecutor keySerializingExecutor = new WrappingKeySerializingExecutor(sameThreadExecutor()); context.checking(new Expectations() {{ oneOf(task).get(); will(throwException(new ArithmeticException("Expected as part of test"))); }}); assertThat(keySerializingExecutor.submit("key", task), resultsInException(ArithmeticException.class)); }
@Override public <T> ListenableFuture<T> submit(String key, CheckedSupplier<T, Exception> task) { if (shutdown) { throw new RejectedExecutionException("WrappingKeySerializingExecutor already shut down"); } SettableFuture<T> taskFinishedFuture = SettableFuture.create(); Runnable taskRunner = createFutureSettingTaskRunner(task, taskFinishedFuture); enqueueOrRunTask(taskRunner, getQueueForKey(key)); return taskFinishedFuture; }
WrappingKeySerializingExecutor implements KeySerializingExecutor { @Override public <T> ListenableFuture<T> submit(String key, CheckedSupplier<T, Exception> task) { if (shutdown) { throw new RejectedExecutionException("WrappingKeySerializingExecutor already shut down"); } SettableFuture<T> taskFinishedFuture = SettableFuture.create(); Runnable taskRunner = createFutureSettingTaskRunner(task, taskFinishedFuture); enqueueOrRunTask(taskRunner, getQueueForKey(key)); return taskFinishedFuture; } }
WrappingKeySerializingExecutor implements KeySerializingExecutor { @Override public <T> ListenableFuture<T> submit(String key, CheckedSupplier<T, Exception> task) { if (shutdown) { throw new RejectedExecutionException("WrappingKeySerializingExecutor already shut down"); } SettableFuture<T> taskFinishedFuture = SettableFuture.create(); Runnable taskRunner = createFutureSettingTaskRunner(task, taskFinishedFuture); enqueueOrRunTask(taskRunner, getQueueForKey(key)); return taskFinishedFuture; } WrappingKeySerializingExecutor(ExecutorService executorService); }
WrappingKeySerializingExecutor implements KeySerializingExecutor { @Override public <T> ListenableFuture<T> submit(String key, CheckedSupplier<T, Exception> task) { if (shutdown) { throw new RejectedExecutionException("WrappingKeySerializingExecutor already shut down"); } SettableFuture<T> taskFinishedFuture = SettableFuture.create(); Runnable taskRunner = createFutureSettingTaskRunner(task, taskFinishedFuture); enqueueOrRunTask(taskRunner, getQueueForKey(key)); return taskFinishedFuture; } WrappingKeySerializingExecutor(ExecutorService executorService); @Override ListenableFuture<T> submit(String key, CheckedSupplier<T, Exception> task); @Override void shutdownAndAwaitTermination(long timeout, TimeUnit unit); }
WrappingKeySerializingExecutor implements KeySerializingExecutor { @Override public <T> ListenableFuture<T> submit(String key, CheckedSupplier<T, Exception> task) { if (shutdown) { throw new RejectedExecutionException("WrappingKeySerializingExecutor already shut down"); } SettableFuture<T> taskFinishedFuture = SettableFuture.create(); Runnable taskRunner = createFutureSettingTaskRunner(task, taskFinishedFuture); enqueueOrRunTask(taskRunner, getQueueForKey(key)); return taskFinishedFuture; } WrappingKeySerializingExecutor(ExecutorService executorService); @Override ListenableFuture<T> submit(String key, CheckedSupplier<T, Exception> task); @Override void shutdownAndAwaitTermination(long timeout, TimeUnit unit);  }