Refactor to support a separate SPI package

This commit includes a large refactoring with various improvements, but generally is aimed towards having a separate SPI package that provides access to most of the Failsafe internals such that it's easy to implement policies. The changes in this commit include:

- Separating the execution types into separate public and internal interfaces
- Creating an spi package and putting the various internal extensible classes in there
- Improved the way internal classes interact with FailsafeFuture

Fixes #292

Author: jhalterman

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

@@ -1,88 +1,26 @@
-/*
- * Copyright 2016 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License
- */
 package net.jodah.failsafe;
 
-import net.jodah.failsafe.internal.util.Assert;
-import net.jodah.failsafe.util.concurrent.Scheduler;
-
-import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionException;
-import java.util.concurrent.Future;
-import java.util.function.Function;
 
 /**
- * Tracks asynchronous executions and handles failures according to one or more {@link Policy policies}. Execution
- * results must be explicitly recorded via one of the {@code record} methods.
+ * Allows asynchronous executions to record their results or complete an execution.
  *
  * @param <R> result type
  * @author Jonathan Halterman
  */
-public final class AsyncExecution<R> extends AbstractExecution<R> {
-  // Cross-attempt state --
-  // The outer-most function that executions begin with
-  private Function<AsyncExecution<R>, CompletableFuture<ExecutionResult>> outerFn;
-  private final boolean asyncExecution;
-  final FailsafeFuture<R> future;
-
-  // Per-attempt state --
-  // Whether a result has been explicitly recorded
-  volatile boolean recordCalled;
-  // The future for the thread that the innerFn is running in
-  volatile Future<?> innerFuture;
-  // Whether a policy executor completed post execution
-  private final boolean[] policyPostExecuted;
-
-  AsyncExecution(List<Policy<R>> policies, Scheduler scheduler, FailsafeFuture<R> future, boolean asyncExecution,
-    Function<AsyncExecution<R>, CompletableFuture<ExecutionResult>> innerFn) {
-    super(policies, scheduler);
-    this.future = future;
-    this.asyncExecution = asyncExecution;
-    this.policyPostExecuted = new boolean[policyExecutors.size()];
-
-    outerFn = asyncExecution ? Functions.toExecutionAware(innerFn) : innerFn;
-    outerFn = Functions.toAsync(outerFn, scheduler);
-
-    for (PolicyExecutor<R, ? extends Policy<R>> policyExecutor : policyExecutors)
-      outerFn = policyExecutor.applyAsync(outerFn, scheduler, this.future);
-  }
-
-  private AsyncExecution(AsyncExecution<R> execution) {
-    super(execution);
-    outerFn = execution.outerFn;
-    future = execution.future;
-    asyncExecution = execution.asyncExecution;
-    policyPostExecuted = new boolean[policyExecutors.size()];
-  }
-
+public interface AsyncExecution<R> extends ExecutionContext<R> {
   /**
    * Completes the execution and the associated {@code CompletableFuture}.
    *
-   * @throws IllegalStateException if the execution is already complete
+   * @throws IllegalStateException if the execution is already recorded or complete
    */
-  public void complete() {
-    Assert.state(!recordCalled, "The most recent execution has already been recorded");
-    recordCalled = true;
+  void complete();
 
-    // Guard against race with a timeout expiring
-    synchronized (future) {
-      ExecutionResult result = this.result != null ? this.result : ExecutionResult.NONE;
-      complete(postExecute(result), null);
-    }
-  }
+  /**
+   * Returns whether the execution is complete or if it can be retried. An execution is considered complete only when
+   * all configured policies consider the execution complete.
+   */
+  boolean isComplete();
 
   /**
    * Records an execution {@code result} or {@code failure} which triggers failure handling, if needed, by the
@@ -91,78 +29,21 @@ public void complete() {
    *
    * @throws IllegalStateException if the most recent execution was already recorded or the execution is complete
    */
-  public void record(R result, Throwable failure) {
-    Assert.state(!recordCalled, "The most recent execution has already been recorded");
-    recordCalled = true;
-
-    // Guard against race with a timeout expiring
-    synchronized (future) {
-      if (!attemptRecorded) {
-        ExecutionResult er = new ExecutionResult(result, failure).withDelay(delayNanos);
-        record(er);
-      }
-
-      // Proceed with handling the recorded result
-      executeAsync();
-    }
-  }
+  void record(R result, Throwable failure);
 
   /**
    * Records an execution {@code result} which triggers failure handling, if needed, by the configured policies. If
    * policy handling is not possible or already complete, the resulting {@link CompletableFuture} is completed.
    *
    * @throws IllegalStateException if the most recent execution was already recorded or the execution is complete
    */
-  public void recordResult(R result) {
-    record(result, null);
-  }
+  void recordResult(R result);
 
   /**
    * Records an execution {@code failure} which triggers failure handling, if needed, by the configured policies. If
    * policy handling is not possible or already complete, the resulting {@link CompletableFuture} is completed.
    *
    * @throws IllegalStateException if the most recent execution was already recorded or the execution is complete
    */
-  public void recordFailure(Throwable failure) {
-    record(null, failure);
-  }
-
-  /**
-   * Performs an asynchronous execution.
-   */
-  void executeAsync() {
-    outerFn.apply(this).whenComplete(this::complete);
-  }
-
-  private void complete(ExecutionResult result, Throwable error) {
-    if (result == null && error == null)
-      return;
-
-    completed = true;
-    if (!future.isDone()) {
-      if (result != null)
-        future.completeResult(result);
-      else {
-        if (error instanceof CompletionException)
-          error = error.getCause();
-        future.completeResult(ExecutionResult.failure(error));
-      }
-    }
-  }
-
-  synchronized void setPostExecuted(int policyIndex) {
-    policyPostExecuted[policyIndex] = true;
-  }
-
-  synchronized boolean isPostExecuted(int policyIndex) {
-    return policyPostExecuted[policyIndex];
-  }
-
-  boolean isAsyncExecution() {
-    return asyncExecution;
-  }
-
-  AsyncExecution<R> copy() {
-    return new AsyncExecution<>(this);
-  }
-}
+  void recordFailure(Throwable failure);
+}