Added improvements and tests around alternative execution results

Author: jhalterman

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

@@ -16,6 +16,7 @@
 package net.jodah.failsafe;
 
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * The result of an execution. Immutable.
@@ -25,6 +26,8 @@
  * @author Jonathan Halterman
  */
 public class ExecutionResult {
+  static final CompletableFuture<ExecutionResult> NULL_FUTURE = CompletableFuture.completedFuture(null);
+
   /** An execution that was completed with a non-result */
   static final ExecutionResult NONE = new ExecutionResult(null, null, true, 0, true, true, true);
 

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

@@ -66,6 +66,9 @@ protected Supplier<ExecutionResult> supply(Supplier<ExecutionResult> supplier, S
   protected Supplier<CompletableFuture<ExecutionResult>> supplyAsync(
     Supplier<CompletableFuture<ExecutionResult>> supplier, Scheduler scheduler, FailsafeFuture<Object> future) {
     return () -> supplier.get().thenCompose(result -> {
+      if (result == null)
+        return ExecutionResult.NULL_FUTURE;
+
       CompletableFuture<ExecutionResult> promise = new CompletableFuture<>();
       if (executionCancelled()) {
         promise.complete(result);

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

@@ -29,8 +29,6 @@
  * @author Jonathan Halterman
  */
 final class Functions {
-  private static final CompletableFuture<ExecutionResult> NULL_FUTURE = CompletableFuture.completedFuture(null);
-
   interface SettableSupplier<T> extends Supplier<T> {
     void set(T value);
   }
@@ -136,7 +134,7 @@ public synchronized CompletableFuture<ExecutionResult> get() {
         }
 
         // Result will be provided later via AsyncExecution.complete
-        return NULL_FUTURE;
+        return ExecutionResult.NULL_FUTURE;
       }
     };
   }
@@ -199,7 +197,7 @@ static <T> Supplier<CompletableFuture<ExecutionResult>> getPromiseOfStageExecuti
       }
 
       // Result will be provided later via AsyncExecution.complete
-      return NULL_FUTURE;
+      return ExecutionResult.NULL_FUTURE;
     };
   }
 

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

@@ -31,8 +31,7 @@
 public abstract class PolicyExecutor<P extends Policy> {
   protected final P policy;
   protected final AbstractExecution execution;
-  /* Index of the policy relative to other policies in a composition, inner-most first */
-  int policyIndex;
+  /* Index of the policy relative to other policies in a composition, inner-most first */ int policyIndex;
 
   protected PolicyExecutor(P policy, AbstractExecution execution) {
     this.policy = policy;
@@ -53,8 +52,11 @@ protected ExecutionResult preExecute() {
   protected Supplier<ExecutionResult> supply(Supplier<ExecutionResult> supplier, Scheduler scheduler) {
     return () -> {
       ExecutionResult result = preExecute();
-      if (result != null)
+      if (result != null) {
+        // Still need to preExecute when returning an alternative result before making it to the terminal Supplier
+        execution.preExecute();
         return result;
+      }
 
       return postExecute(supplier.get());
     };
@@ -64,6 +66,7 @@ protected Supplier<ExecutionResult> supply(Supplier<ExecutionResult> supplier, S
    * Performs synchronous post-execution handling for a {@code result}.
    */
   protected ExecutionResult postExecute(ExecutionResult result) {
+    execution.recordAttempt();
     if (isFailure(result)) {
       result = onFailure(result.withFailure());
       callFailureListener(result);
@@ -77,16 +80,23 @@ protected ExecutionResult postExecute(ExecutionResult result) {
   }
 
   /**
-   * Performs an async execution by calling pre-execute else calling the supplier and doing a post-execute.
+   * Performs an async execution by calling pre-execute else calling the supplier and doing a post-execute. Implementors
+   * must handle a null result from a supplier, which indicates that an async execution has occurred, a result will come
+   * later, and postExecute handling should not be performed.
    */
   protected Supplier<CompletableFuture<ExecutionResult>> supplyAsync(
     Supplier<CompletableFuture<ExecutionResult>> supplier, Scheduler scheduler, FailsafeFuture<Object> future) {
     return () -> {
       ExecutionResult result = preExecute();
-      if (result != null)
+      if (result != null) {
+        // Still need to preExecute when returning an alternative result before making it to the terminal Supplier
+        execution.preExecute();
         return CompletableFuture.completedFuture(result);
+      }
 
-      return supplier.get().thenCompose(s -> postExecuteAsync(s, scheduler, future));
+      return supplier.get().thenCompose(r -> {
+        return r == null ? ExecutionResult.NULL_FUTURE : postExecuteAsync(r, scheduler, future);
+      });
     };
   }
 

src/test/java/net/jodah/failsafe/Testing.java

@@ -16,10 +16,7 @@
 package net.jodah.failsafe;
 
 import net.jodah.failsafe.event.ExecutionCompletedEvent;
-import net.jodah.failsafe.function.AsyncRunnable;
-import net.jodah.failsafe.function.CheckedConsumer;
-import net.jodah.failsafe.function.CheckedRunnable;
-import net.jodah.failsafe.function.CheckedSupplier;
+import net.jodah.failsafe.function.*;
 import net.jodah.failsafe.internal.CircuitBreakerInternals;
 import net.jodah.failsafe.internal.CircuitState;
 
@@ -35,6 +32,8 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
+import static org.testng.Assert.assertEquals;
+
 /**
  * Utilities to to assist with testing.
  */
@@ -185,6 +184,7 @@ public static <T> T unwrapExceptions(CheckedSupplier<T> supplier) {
       sneakyThrow(e.getCause() == null ? e : e.getCause());
       return null;
     } catch (RuntimeException | Error e) {
+      e.printStackTrace();
       throw e;
     } catch (Throwable t) {
       throw new RuntimeException(t);
@@ -329,23 +329,56 @@ public static void testSyncAndAsync(FailsafeExecutor<?> failsafe, Runnable given
   @SafeVarargs
   public static void testAsyncExecution(FailsafeExecutor<?> failsafe, AsyncRunnable when,
     Consumer<ExecutionCompletedEvent<?>> then, Class<? extends Throwable>... expectedExceptions) {
+    AsyncSupplier supplier = ex -> {
+      when.run(ex);
+      return null;
+    };
+    testAsyncExecution(failsafe, supplier, then, null, expectedExceptions);
+  }
+
+  @SafeVarargs
+  public static void testAsyncExecution(FailsafeExecutor<?> failsafe, AsyncRunnable when,
+    Consumer<ExecutionCompletedEvent<?>> then, Object expectedResult,
+    Class<? extends Throwable>... expectedExceptions) {
+    AsyncSupplier supplier = ex -> {
+      when.run(ex);
+      return null;
+    };
+    testAsyncExecution(failsafe, supplier, then, expectedResult, expectedExceptions);
+  }
+
+  @SafeVarargs
+  public static <T> void testAsyncExecution(FailsafeExecutor<T> failsafe, AsyncSupplier<T> when,
+    Consumer<ExecutionCompletedEvent<?>> then, Class<? extends Throwable>... expectedExceptions) {
+    testAsyncExecution(failsafe, when, then, null, expectedExceptions);
+  }
+
+  @SafeVarargs
+  public static <T> void testAsyncExecution(FailsafeExecutor<T> failsafe, AsyncSupplier<T> when,
+    Consumer<ExecutionCompletedEvent<?>> then, Object expectedResult,
+    Class<? extends Throwable>... expectedExceptions) {
     AtomicReference<ExecutionCompletedEvent<?>> completedEventRef = new AtomicReference<>();
     CheckedConsumer<ExecutionCompletedEvent<?>> setCompletedEventFn = completedEventRef::set;
     Runnable postTestFn = () -> {
+      ExecutionCompletedEvent<?> completedEvent = completedEventRef.get();
       if (expectedExceptions.length > 0)
-        Asserts.assertMatches(completedEventRef.get().getFailure(), Arrays.asList(expectedExceptions));
+        Asserts.assertMatches(completedEvent.getFailure(), Arrays.asList(expectedExceptions));
+      else
+        assertEquals(completedEvent.getResult(), expectedResult);
       then.accept(completedEventRef.get());
     };
 
     // Async test
     System.out.println("\nRunning async execution test");
     if (expectedExceptions.length == 0) {
-      Testing.unwrapExceptions(() -> failsafe.onComplete(setCompletedEventFn::accept).runAsyncExecution(when).get());
+      Object result = Testing.unwrapExceptions(
+        () -> failsafe.onComplete(setCompletedEventFn::accept).getAsyncExecution(when).get());
+      assertEquals(result, expectedResult);
     } else {
       List<Class<? extends Throwable>> expected = new LinkedList<>();
       Collections.addAll(expected, expectedExceptions);
       expected.add(0, ExecutionException.class);
-      Asserts.assertThrows(() -> failsafe.onComplete(setCompletedEventFn::accept).runAsyncExecution(when).get(),
+      Asserts.assertThrows(() -> failsafe.onComplete(setCompletedEventFn::accept).getAsyncExecution(when).get(),
         expected);
     }
     postTestFn.run();

src/test/java/net/jodah/failsafe/functional/AlternativeResultTest.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright 2021 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.functional;
+
+import net.jodah.failsafe.*;
+import net.jodah.failsafe.Testing.Stats;
+import org.testng.annotations.Test;
+
+import static net.jodah.failsafe.Testing.*;
+import static org.testng.Assert.assertEquals;
+
+/**
+ * Tests scenarios where a PolicyExecutor's preExecute provides an alternative result, which prevents the user's
+ * Supplier from being called. This occurs when a CircuitBreaker is open.
+ */
+@Test
+public class AlternativeResultTest {
+  public void testRejectedSyncAndAsync() {
+    Stats rpStats = new Stats();
+    Stats cbStats = new Stats();
+    RetryPolicy<Object> rp = withStats(new RetryPolicy<>().withMaxAttempts(7), rpStats, true);
+    CircuitBreaker<Object> cb = withStats(new CircuitBreaker<>().withFailureThreshold(3), cbStats, true);
+
+    testSyncAndAsync(Failsafe.with(rp, cb), () -> {
+      rpStats.reset();
+      cbStats.reset();
+      cb.close();
+    }, () -> {
+      System.out.println("Executing");
+      throw new Exception();
+    }, e -> {
+      assertEquals(e.getAttemptCount(), 7);
+      assertEquals(e.getExecutionCount(), 3);
+      assertEquals(rpStats.failedAttemptCount, 7);
+      assertEquals(rpStats.retryCount, 6);
+      assertEquals(cb.getExecutionCount(), 3);
+      assertEquals(cb.getFailureCount(), 3);
+    }, CircuitBreakerOpenException.class);
+  }
+
+  public void testRejectedAsyncExecutionWithRetry() {
+    Stats rpStats = new Stats();
+    Stats cbStats = new Stats();
+    RetryPolicy<Object> rp = withStats(new RetryPolicy<>().withMaxAttempts(7), rpStats, true);
+    CircuitBreaker<Object> cb = withStats(new CircuitBreaker<>().withFailureThreshold(3), cbStats, true);
+
+    // Test with retryOn()
+    testAsyncExecution(Failsafe.with(rp, cb), ex -> {
+      runAsync(() -> {
+        System.out.println("Executing");
+        ex.retryOn(new IllegalStateException());
+      });
+    }, e -> {
+      assertEquals(e.getAttemptCount(), 7);
+      assertEquals(e.getExecutionCount(), 3);
+      assertEquals(rpStats.failedAttemptCount, 7);
+      assertEquals(rpStats.retryCount, 6);
+      assertEquals(cb.getExecutionCount(), 3);
+      assertEquals(cb.getFailureCount(), 3);
+    }, CircuitBreakerOpenException.class);
+  }
+
+  @Test(enabled = false)
+  public void testRejectedAsyncExecutionWithCompleteAndRetry() {
+    Stats rpStats = new Stats();
+    Stats cbStats = new Stats();
+    RetryPolicy<Object> rp = withStats(new RetryPolicy<>().withMaxAttempts(7), rpStats, true);
+    CircuitBreaker<Object> cb = withStats(new CircuitBreaker<>().withFailureThreshold(3), cbStats, true);
+
+    // Test with complete() and retry()
+    rpStats.reset();
+    cbStats.reset();
+    cb.close();
+    testAsyncExecution(Failsafe.with(rp, cb), ex -> {
+      runAsync(() -> {
+        System.out.println("Executing");
+        if (!ex.complete(null, new IllegalStateException())) {
+          ex.retry();
+        }
+      });
+    }, e -> {
+      assertEquals(e.getAttemptCount(), 7);
+      assertEquals(e.getExecutionCount(), 3);
+      assertEquals(rpStats.failedAttemptCount, 7);
+      assertEquals(rpStats.retryCount, 6);
+      assertEquals(cb.getExecutionCount(), 3);
+      assertEquals(cb.getFailureCount(), 3);
+    }, CircuitBreakerOpenException.class);
+  }
+}

src/test/java/net/jodah/failsafe/functional/AsyncExecutionTest.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2021 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.functional;
+
+import net.jodah.failsafe.*;
+import net.jodah.failsafe.Testing.Stats;
+import org.testng.annotations.Test;
+
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+
+import static net.jodah.failsafe.Testing.*;
+import static org.testng.Assert.assertEquals;
+
+/**
+ * Tests AsyncExecution scenarios.
+ */
+@Test
+public class AsyncExecutionTest {
+  public void testAsyncExecWithPassthroughPolicies() {
+    Stats rpStats = new Stats();
+    RetryPolicy<Object> rp = withStats(new RetryPolicy<>().withMaxRetries(3), rpStats, true);
+    // Passthrough policy that should allow async execution results through
+    Fallback<Object> fb = Fallback.<Object>of("test").handleIf((r, f) -> false);
+    Timeout<Object> timeout = Timeout.of(Duration.ofMinutes(1));
+    AtomicInteger counter = new AtomicInteger();
+
+    Consumer<FailsafeExecutor<Object>> test = failsafe -> testAsyncExecution(failsafe, ex -> {
+      runAsync(() -> {
+        System.out.println("Executing");
+        if (counter.getAndIncrement() < 3)
+          ex.retryOn(new IllegalStateException());
+        else
+          ex.complete("done");
+      });
+    }, e -> {
+      assertEquals(e.getAttemptCount(), 4);
+      assertEquals(e.getExecutionCount(), 4);
+      assertEquals(rpStats.failedAttemptCount, 3);
+      assertEquals(rpStats.retryCount, 3);
+    }, "done");
+
+    // Test RetryPolicy, Fallback
+    test.accept(Failsafe.with(rp, fb));
+
+    // Test RetryPolicy, Timeout
+    rpStats.reset();
+    counter.set(0);
+    test.accept(Failsafe.with(rp, timeout));
+  }
+}