@Test public void submitsTasksOnceEachToTheSuppliedExecutorService() throws Exception { KeySerializingExecutor keySerializingExecutor = new WrappingKeySerializingExecutor(executorService); context.checking(new Expectations() {{ allowSubmitOrExecuteOnce(context, executorService); }}); keySerializingExecutor.submit("key", task); }
@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);  }