@Test void shouldReportRejectionFromWorkerPoolForDelayedActions() { final RejectedExecutionException exception = new RejectedExecutionException("Too lazy"); doThrow(exception).when(workerPool).execute(any()); final SafeFuture<Void> result = asyncRunner.runAfterDelay(() -> {}, 100, TimeUnit.MILLISECONDS); final ArgumentCaptor<Runnable> taskCaptor = ArgumentCaptor.forClass(Runnable.class); verify(scheduler).schedule(taskCaptor.capture(), eq(100L), eq(TimeUnit.MILLISECONDS)); taskCaptor.getValue().run(); SafeFutureAssert.assertThatSafeFuture(result).isCompletedExceptionallyWith(exception); }
@Override @SuppressWarnings("FutureReturnValueIgnored") public <U> SafeFuture<U> runAfterDelay( final Supplier<SafeFuture<U>> action, final long delayAmount, final TimeUnit delayUnit) { if (shutdown.get()) { LOG.debug("Ignoring async task because shutdown is in progress"); return new SafeFuture<>(); } final SafeFuture<U> result = new SafeFuture<>(); try { scheduler.schedule( () -> { try { workerPool.execute(createRunnableForAction(action, result)); } catch (final Throwable t) { handleExecutorError(result, t); } }, delayAmount, delayUnit); } catch (final Throwable t) { handleExecutorError(result, t); } return result; }
ScheduledExecutorAsyncRunner implements AsyncRunner { @Override @SuppressWarnings("FutureReturnValueIgnored") public <U> SafeFuture<U> runAfterDelay( final Supplier<SafeFuture<U>> action, final long delayAmount, final TimeUnit delayUnit) { if (shutdown.get()) { LOG.debug("Ignoring async task because shutdown is in progress"); return new SafeFuture<>(); } final SafeFuture<U> result = new SafeFuture<>(); try { scheduler.schedule( () -> { try { workerPool.execute(createRunnableForAction(action, result)); } catch (final Throwable t) { handleExecutorError(result, t); } }, delayAmount, delayUnit); } catch (final Throwable t) { handleExecutorError(result, t); } return result; } }
ScheduledExecutorAsyncRunner implements AsyncRunner { @Override @SuppressWarnings("FutureReturnValueIgnored") public <U> SafeFuture<U> runAfterDelay( final Supplier<SafeFuture<U>> action, final long delayAmount, final TimeUnit delayUnit) { if (shutdown.get()) { LOG.debug("Ignoring async task because shutdown is in progress"); return new SafeFuture<>(); } final SafeFuture<U> result = new SafeFuture<>(); try { scheduler.schedule( () -> { try { workerPool.execute(createRunnableForAction(action, result)); } catch (final Throwable t) { handleExecutorError(result, t); } }, delayAmount, delayUnit); } catch (final Throwable t) { handleExecutorError(result, t); } return result; } ScheduledExecutorAsyncRunner(
      final ScheduledExecutorService scheduler, final ExecutorService workerPool); }
ScheduledExecutorAsyncRunner implements AsyncRunner { @Override @SuppressWarnings("FutureReturnValueIgnored") public <U> SafeFuture<U> runAfterDelay( final Supplier<SafeFuture<U>> action, final long delayAmount, final TimeUnit delayUnit) { if (shutdown.get()) { LOG.debug("Ignoring async task because shutdown is in progress"); return new SafeFuture<>(); } final SafeFuture<U> result = new SafeFuture<>(); try { scheduler.schedule( () -> { try { workerPool.execute(createRunnableForAction(action, result)); } catch (final Throwable t) { handleExecutorError(result, t); } }, delayAmount, delayUnit); } catch (final Throwable t) { handleExecutorError(result, t); } return result; } ScheduledExecutorAsyncRunner(
      final ScheduledExecutorService scheduler, final ExecutorService workerPool); static AsyncRunner create(
      final String name,
      final int maxThreads,
      final MetricTrackingExecutorFactory executorFactory); @Override SafeFuture<U> runAsync(final Supplier<SafeFuture<U>> action); @Override @SuppressWarnings("FutureReturnValueIgnored") SafeFuture<U> runAfterDelay(
      final Supplier<SafeFuture<U>> action, final long delayAmount, final TimeUnit delayUnit); @Override void shutdown(); }
ScheduledExecutorAsyncRunner implements AsyncRunner { @Override @SuppressWarnings("FutureReturnValueIgnored") public <U> SafeFuture<U> runAfterDelay( final Supplier<SafeFuture<U>> action, final long delayAmount, final TimeUnit delayUnit) { if (shutdown.get()) { LOG.debug("Ignoring async task because shutdown is in progress"); return new SafeFuture<>(); } final SafeFuture<U> result = new SafeFuture<>(); try { scheduler.schedule( () -> { try { workerPool.execute(createRunnableForAction(action, result)); } catch (final Throwable t) { handleExecutorError(result, t); } }, delayAmount, delayUnit); } catch (final Throwable t) { handleExecutorError(result, t); } return result; } ScheduledExecutorAsyncRunner(
      final ScheduledExecutorService scheduler, final ExecutorService workerPool); static AsyncRunner create(
      final String name,
      final int maxThreads,
      final MetricTrackingExecutorFactory executorFactory); @Override SafeFuture<U> runAsync(final Supplier<SafeFuture<U>> action); @Override @SuppressWarnings("FutureReturnValueIgnored") SafeFuture<U> runAfterDelay(
      final Supplier<SafeFuture<U>> action, final long delayAmount, final TimeUnit delayUnit); @Override void shutdown();  }