Use Executor for sync executions

Related #221

Author: jhalterman

src/main/java/net/jodah/failsafe/FailsafeExecutor.java

@@ -43,6 +43,7 @@
  */
 public class FailsafeExecutor<R> extends PolicyListeners<FailsafeExecutor<R>, R> {
   private Scheduler scheduler = Scheduler.DEFAULT;
+  private Executor executor;
   /** Policies sorted outer-most first */
   final List<? extends Policy<R>> policies;
   private EventHandler<R> completeHandler;
@@ -134,7 +135,7 @@ public <T extends R> T get(ContextualSupplier<T, T> supplier) {
    * @throws RejectedExecutionException if the {@code supplier} cannot be scheduled for execution
    */
   public <T extends R> CompletableFuture<T> getAsync(CheckedSupplier<T> supplier) {
-    return callAsync(future -> getPromise(toCtxSupplier(supplier)), false);
+    return callAsync(future -> getPromise(toCtxSupplier(supplier), executor), false);
   }
 
   /**
@@ -153,7 +154,7 @@ public <T extends R> CompletableFuture<T> getAsync(CheckedSupplier<T> supplier)
    * @throws RejectedExecutionException if the {@code supplier} cannot be scheduled for execution
    */
   public <T extends R> CompletableFuture<T> getAsync(ContextualSupplier<T, T> supplier) {
-    return callAsync(future -> getPromise(supplier), false);
+    return callAsync(future -> getPromise(supplier, executor), false);
   }
 
   /**
@@ -178,7 +179,7 @@ public <T extends R> CompletableFuture<T> getAsync(ContextualSupplier<T, T> supp
    * @throws RejectedExecutionException if the {@code supplier} cannot be scheduled for execution
    */
   public <T extends R> CompletableFuture<T> getAsyncExecution(AsyncRunnable<T> runnable) {
-    return callAsync(future -> getPromiseExecution(runnable), true);
+    return callAsync(future -> getPromiseExecution(runnable, executor), true);
   }
 
   /**
@@ -199,7 +200,7 @@ public <T extends R> CompletableFuture<T> getAsyncExecution(AsyncRunnable<T> run
    * @throws RejectedExecutionException if the {@code supplier} cannot be scheduled for execution
    */
   public <T extends R> CompletableFuture<T> getStageAsync(CheckedSupplier<? extends CompletionStage<T>> supplier) {
-    return callAsync(future -> getPromiseOfStage(toCtxSupplier(supplier), future), false);
+    return callAsync(future -> getPromiseOfStage(toCtxSupplier(supplier), future, executor), false);
   }
 
   /**
@@ -221,7 +222,7 @@ public <T extends R> CompletableFuture<T> getStageAsync(CheckedSupplier<? extend
    */
   public <T extends R> CompletableFuture<T> getStageAsync(
     ContextualSupplier<T, ? extends CompletionStage<T>> supplier) {
-    return callAsync(future -> getPromiseOfStage(supplier, future), false);
+    return callAsync(future -> getPromiseOfStage(supplier, future, executor), false);
   }
 
   /**
@@ -249,7 +250,7 @@ public <T extends R> CompletableFuture<T> getStageAsync(
    */
   public <T extends R> CompletableFuture<T> getStageAsyncExecution(
     AsyncSupplier<T, ? extends CompletionStage<T>> supplier) {
-    return callAsync(future -> getPromiseOfStageExecution(supplier, future), true);
+    return callAsync(future -> getPromiseOfStageExecution(supplier, future, executor), true);
   }
 
   /**
@@ -293,7 +294,7 @@ public void run(ContextualRunnable<Void> runnable) {
    * @throws RejectedExecutionException if the {@code runnable} cannot be scheduled for execution
    */
   public CompletableFuture<Void> runAsync(CheckedRunnable runnable) {
-    return callAsync(future -> getPromise(toCtxSupplier(runnable)), false);
+    return callAsync(future -> getPromise(toCtxSupplier(runnable), executor), false);
   }
 
   /**
@@ -311,7 +312,7 @@ public CompletableFuture<Void> runAsync(CheckedRunnable runnable) {
    * @throws RejectedExecutionException if the {@code runnable} cannot be scheduled for execution
    */
   public CompletableFuture<Void> runAsync(ContextualRunnable<Void> runnable) {
-    return callAsync(future -> getPromise(toCtxSupplier(runnable)), false);
+    return callAsync(future -> getPromise(toCtxSupplier(runnable), executor), false);
   }
 
   /**
@@ -336,7 +337,7 @@ public CompletableFuture<Void> runAsync(ContextualRunnable<Void> runnable) {
    * @throws RejectedExecutionException if the {@code runnable} cannot be scheduled for execution
    */
   public CompletableFuture<Void> runAsyncExecution(AsyncRunnable<Void> runnable) {
-    return callAsync(future -> getPromiseExecution(runnable), true);
+    return callAsync(future -> getPromiseExecution(runnable, executor), true);
   }
 
   /**
@@ -372,33 +373,46 @@ public FailsafeExecutor<R> onSuccess(CheckedConsumer<ExecutionCompletedEvent<R>>
   }
 
   /**
-   * Configures the {@code executor} to use for performing asynchronous executions and listener callbacks.
+   * Configures the {@code scheduledExecutorService} to use for performing asynchronous executions and listener
+   * callbacks.
    * <p>
-   * Note: The {@code executor} should have a core pool size of at least 2 in order for {@link Timeout timeouts} to
-   * work.
+   * Note: The {@code scheduledExecutorService} should have a core pool size of at least 2 in order for {@link Timeout
+   * timeouts} to work.
    * </p>
    *
-   * @throws NullPointerException if {@code executor} is null
-   * @throws IllegalArgumentException if the {@code executor} has a core pool size of less than 2
+   * @throws NullPointerException if {@code scheduledExecutorService} is null
+   * @throws IllegalArgumentException if the {@code scheduledExecutorService} has a core pool size of less than 2
    */
-  public FailsafeExecutor<R> with(ScheduledExecutorService executor) {
-    this.scheduler = Scheduler.of(executor);
+  public FailsafeExecutor<R> with(ScheduledExecutorService scheduledExecutorService) {
+    this.scheduler = Scheduler.of(Assert.notNull(scheduledExecutorService, "scheduledExecutorService"));
     return this;
   }
 
   /**
-   * Configures the {@code executor} to use for performing asynchronous executions and listener callbacks. For
-   * executions that require a delay, an internal ScheduledExecutorService will be used for the delay, then the {@code
-   * executor} will be used for actual execution.
+   * Configures the {@code executorService} to use for performing asynchronous executions and listener callbacks. For
+   * async executions that require a delay, an internal ScheduledExecutorService will be used for the delay, then the
+   * {@code executorService} will be used for actual execution.
    * <p>
-   * Note: The {@code executor} should have a core pool size or parallelism of at least 2 in order for {@link Timeout
-   * timeouts} to work.
+   * Note: The {@code executorService} should have a core pool size or parallelism of at least 2 in order for {@link
+   * Timeout timeouts} to work.
    * </p>
    *
+   * @throws NullPointerException if {@code executorService} is null
+   */
+  public FailsafeExecutor<R> with(ExecutorService executorService) {
+    this.scheduler = Scheduler.of(Assert.notNull(executorService, "executorService"));
+    return this;
+  }
+
+  /**
+   * Configures the {@code executor} to use as a wrapper around executions. The {@code executor} is responsible for
+   * propagating executions. Executions that normally return a result, such as {@link #get(CheckedSupplier)} will return
+   * {@code null} since the {@link Executor} interface does not support results.
+   *
    * @throws NullPointerException if {@code executor} is null
    */
   public FailsafeExecutor<R> with(Executor executor) {
-    this.scheduler = Scheduler.of(executor);
+    this.executor = Assert.notNull(executor, "executor");
     return this;
   }
 
@@ -422,7 +436,7 @@ public FailsafeExecutor<R> with(Scheduler scheduler) {
    */
   @SuppressWarnings({ "unchecked", "rawtypes" })
   private <T> T call(ContextualSupplier<T, T> innerSupplier) {
-    SyncExecutionImpl<T> execution = new SyncExecutionImpl(this, scheduler, Functions.get(innerSupplier));
+    SyncExecutionImpl<T> execution = new SyncExecutionImpl(this, scheduler, Functions.get(innerSupplier, executor));
     ExecutionResult<T> result = execution.executeSync();
     Throwable failure = result.getFailure();
     if (failure != null) {

src/main/java/net/jodah/failsafe/Functions.java

@@ -35,13 +35,15 @@ final class Functions {
    *
    * @param <R> result type
    */
-  static <R> Function<SyncExecutionInternal<R>, ExecutionResult<R>> get(ContextualSupplier<R, R> supplier) {
+  static <R> Function<SyncExecutionInternal<R>, ExecutionResult<R>> get(ContextualSupplier<R, R> supplier,
+    Executor executor) {
+
     return execution -> {
       ExecutionResult<R> result;
       Throwable throwable = null;
       try {
         execution.preExecute();
-        result = ExecutionResult.success(supplier.get(execution));
+        result = ExecutionResult.success(withExecutor(supplier, executor).get(execution));
       } catch (Throwable t) {
         throwable = t;
         result = ExecutionResult.failure(t);
@@ -71,14 +73,14 @@ static <R> Function<SyncExecutionInternal<R>, ExecutionResult<R>> get(Contextual
    * @param <R> result type
    */
   static <R> Function<AsyncExecutionInternal<R>, CompletableFuture<ExecutionResult<R>>> getPromise(
-    ContextualSupplier<R, R> supplier) {
+    ContextualSupplier<R, R> supplier, Executor executor) {
 
     Assert.notNull(supplier, "supplier");
     return execution -> {
       ExecutionResult<R> result;
       try {
         execution.preExecute();
-        result = ExecutionResult.success(supplier.get(execution));
+        result = ExecutionResult.success(withExecutor(supplier, executor).get(execution));
       } catch (Throwable t) {
         result = ExecutionResult.failure(t);
       }
@@ -95,15 +97,15 @@ static <R> Function<AsyncExecutionInternal<R>, CompletableFuture<ExecutionResult
    * @param <R> result type
    */
   static <R> Function<AsyncExecutionInternal<R>, CompletableFuture<ExecutionResult<R>>> getPromiseExecution(
-    AsyncRunnable<R> runnable) {
+    AsyncRunnable<R> runnable, Executor executor) {
 
     Assert.notNull(runnable, "runnable");
     return new Function<AsyncExecutionInternal<R>, CompletableFuture<ExecutionResult<R>>>() {
       @Override
       public synchronized CompletableFuture<ExecutionResult<R>> apply(AsyncExecutionInternal<R> execution) {
         try {
           execution.preExecute();
-          runnable.run(execution);
+          withExecutor(runnable, executor).run(execution);
         } catch (Throwable e) {
           execution.record(null, e);
         }
@@ -122,14 +124,15 @@ public synchronized CompletableFuture<ExecutionResult<R>> apply(AsyncExecutionIn
    */
   @SuppressWarnings("unchecked")
   static <R> Function<AsyncExecutionInternal<R>, CompletableFuture<ExecutionResult<R>>> getPromiseOfStage(
-    ContextualSupplier<R, ? extends CompletionStage<? extends R>> supplier, FailsafeFuture<R> future) {
+    ContextualSupplier<R, ? extends CompletionStage<? extends R>> supplier, FailsafeFuture<R> future,
+    Executor executor) {
 
     Assert.notNull(supplier, "supplier");
     return execution -> {
       CompletableFuture<ExecutionResult<R>> promise = new CompletableFuture<>();
       try {
         execution.preExecute();
-        CompletionStage<? extends R> stage = supplier.get(execution);
+        CompletionStage<? extends R> stage = withExecutor(supplier, executor).get(execution);
 
         // Propagate outer cancellations to the stage
         if (stage instanceof Future)
@@ -160,15 +163,15 @@ static <R> Function<AsyncExecutionInternal<R>, CompletableFuture<ExecutionResult
    */
   @SuppressWarnings("unchecked")
   static <R> Function<AsyncExecutionInternal<R>, CompletableFuture<ExecutionResult<R>>> getPromiseOfStageExecution(
-    AsyncSupplier<R, ? extends CompletionStage<? extends R>> supplier, FailsafeFuture<R> future) {
+    AsyncSupplier<R, ? extends CompletionStage<? extends R>> supplier, FailsafeFuture<R> future, Executor executor) {
 
     Assert.notNull(supplier, "supplier");
     Semaphore asyncFutureLock = new Semaphore(1);
     return execution -> {
       try {
         execution.preExecute();
         asyncFutureLock.acquire();
-        CompletionStage<? extends R> stage = supplier.get(execution);
+        CompletionStage<? extends R> stage = withExecutor(supplier, executor).get(execution);
 
         // Propagate outer cancellations to the stage
         if (stage instanceof Future)
@@ -277,6 +280,52 @@ static <R, T> ContextualSupplier<R, T> toCtxSupplier(CheckedSupplier<T> supplier
     return ctx -> supplier.get();
   }
 
+  static <R, T> ContextualSupplier<R, T> withExecutor(ContextualSupplier<R, T> supplier, Executor executor) {
+    return executor == null ? supplier : ctx -> {
+      executor.execute(() -> {
+        try {
+          supplier.get(ctx);
+        } catch (Throwable e) {
+          handleExecutorThrowablen(e);
+        }
+      });
+      return null;
+    };
+  }
+
+  static <R> AsyncRunnable<R> withExecutor(AsyncRunnable<R> runnable, Executor executor) {
+    return executor == null ? runnable : exec -> {
+      executor.execute(() -> {
+        try {
+          runnable.run(exec);
+        } catch (Throwable e) {
+          handleExecutorThrowablen(e);
+        }
+      });
+    };
+  }
+
+  static <R, T> AsyncSupplier<R, T> withExecutor(AsyncSupplier<R, T> supplier, Executor executor) {
+    return executor == null ? supplier : exec -> {
+      executor.execute(() -> {
+        try {
+          supplier.get(exec);
+        } catch (Throwable e) {
+          handleExecutorThrowablen(e);
+        }
+      });
+      return null;
+    };
+  }
+
+  private static void handleExecutorThrowablen(Throwable e) {
+    if (e instanceof RuntimeException)
+      throw (RuntimeException) e;
+    if (e instanceof Error)
+      throw (Error) e;
+    throw new FailsafeException(e);
+  }
+
   static <T, R> CheckedFunction<T, R> toFn(CheckedConsumer<T> consumer) {
     return t -> {
       consumer.accept(t);

src/main/java/net/jodah/failsafe/internal/util/DelegatingExecutorService.java

@@ -15,8 +15,6 @@
  */
 package net.jodah.failsafe.spi;
 
-import net.jodah.failsafe.internal.util.Assert;
-import net.jodah.failsafe.internal.util.DelegatingExecutorService;
 import net.jodah.failsafe.internal.util.DelegatingScheduler;
 
 import java.util.concurrent.*;
@@ -40,25 +38,16 @@ public interface Scheduler {
   ScheduledFuture<?> schedule(Callable<?> callable, long delay, TimeUnit unit);
 
   /**
-   * Returns a Scheduler adapted from the {@code executor}.
-   *
-   * @throws NullPointerException if {@code executor} is null
+   * Returns a Scheduler adapted from the {@code scheduledExecutorService}.
    */
-  static Scheduler of(final ScheduledExecutorService executor) {
-    Assert.notNull(executor, "executor");
-    return executor::schedule;
+  static Scheduler of(ScheduledExecutorService scheduledExecutorService) {
+    return scheduledExecutorService::schedule;
   }
 
   /**
-   * Returns a Scheduler adapted from the {@code executor}.
-   *
-   * @throws NullPointerException if {@code executor} is null
+   * Returns a Scheduler adapted from the {@code executorService}.
    */
-  static Scheduler of(final Executor executor) {
-    Assert.notNull(executor, "executor");
-    ExecutorService executorService = executor instanceof ExecutorService ?
-      (ExecutorService) executor :
-      new DelegatingExecutorService(executor);
+  static Scheduler of(ExecutorService executorService) {
     return new DelegatingScheduler(executorService);
   }
 }

src/main/java/net/jodah/failsafe/spi/Scheduler.java

@@ -31,7 +31,7 @@
 @Test
 public class AsyncExecutionTest extends Testing {
   Function<AsyncExecutionInternal<Object>, CompletableFuture<ExecutionResult<Object>>> innerFn = Functions.getPromise(
-    ctx -> null);
+    ctx -> null, null);
   ConnectException e = new ConnectException();
   AsyncExecutionInternal<Object> exec;
   FailsafeFuture<Object> future;