Update FallbackExecutor to apply fallback prior to postExecute

Updates FallbackExecutor to apply fallback prior to postExecute. This ensures that any event handling done as part of postExecute takes the final fallback result into account.

Partially addresses #248.

Author: jhalterman

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

@@ -18,6 +18,7 @@
 import net.jodah.failsafe.util.concurrent.Scheduler;
 
 import java.util.concurrent.*;
+import java.util.function.Supplier;
 
 /**
  * A PolicyExecutor that handles failures according to a {@link Fallback}.
@@ -27,47 +28,68 @@ class FallbackExecutor extends PolicyExecutor<Fallback> {
     super(fallback, execution);
   }
 
+  /**
+   * Performs an execution by calling pre-execute else calling the supplier, applying a fallback if it fails, and
+   * calling post-execute.
+   */
   @Override
   @SuppressWarnings("unchecked")
-  protected ExecutionResult onFailure(ExecutionResult result) {
-    try {
-      return policy == Fallback.VOID ?
-        result.withNonResult() :
-        result.withResult(policy.apply(result.getResult(), result.getFailure(), execution.copy()));
-    } catch (Throwable t) {
-      return ExecutionResult.failure(t);
-    }
+  protected Supplier<ExecutionResult> supply(Supplier<ExecutionResult> supplier, Scheduler scheduler) {
+    return () -> {
+      ExecutionResult result = supplier.get();
+      if (isFailure(result)) {
+        try {
+          result = policy == Fallback.VOID ?
+            result.withNonResult() :
+            result.withResult(policy.apply(result.getResult(), result.getFailure(), execution.copy()));
+        } catch (Throwable t) {
+          result = ExecutionResult.failure(t);
+        }
+      }
+
+      return postExecute(result);
+    };
   }
 
+  /**
+   * Performs an async execution by calling pre-execute else calling the supplier and doing a post-execute.
+   */
+  @Override
   @SuppressWarnings("unchecked")
-  protected CompletableFuture<ExecutionResult> onFailureAsync(ExecutionResult result, Scheduler scheduler,
-    FailsafeFuture<Object> future) {
-    CompletableFuture<ExecutionResult> promise = new CompletableFuture<>();
-    Callable<Object> callable = () -> {
-      try {
-        CompletableFuture<Object> fallback = policy.applyStage(result.getResult(), result.getFailure(),
-          execution.copy());
-        fallback.whenComplete((innerResult, failure) -> {
-          if (failure instanceof CompletionException)
-            failure = failure.getCause();
-          ExecutionResult r = failure == null ? result.withResult(innerResult) : ExecutionResult.failure(failure);
-          promise.complete(r);
-        });
-      } catch (Throwable t) {
-        promise.complete(ExecutionResult.failure(t));
-      }
-      return null;
-    };
+  protected Supplier<CompletableFuture<ExecutionResult>> supplyAsync(
+    Supplier<CompletableFuture<ExecutionResult>> supplier, Scheduler scheduler, FailsafeFuture<Object> future) {
+    return () -> supplier.get().thenCompose(result -> {
+      if (isFailure(result)) {
+        CompletableFuture<ExecutionResult> promise = new CompletableFuture<>();
+        Callable<Object> callable = () -> {
+          try {
+            CompletableFuture<Object> fallback = policy.applyStage(result.getResult(), result.getFailure(),
+              execution.copy());
+            fallback.whenComplete((innerResult, failure) -> {
+              if (failure instanceof CompletionException)
+                failure = failure.getCause();
+              ExecutionResult r = failure == null ? result.withResult(innerResult) : ExecutionResult.failure(failure);
+              promise.complete(r);
+            });
+          } catch (Throwable t) {
+            promise.complete(ExecutionResult.failure(t));
+          }
+          return null;
+        };
 
-    try {
-      if (!policy.isAsync())
-        callable.call();
-      else
-        future.inject((Future) scheduler.schedule(callable, result.getWaitNanos(), TimeUnit.NANOSECONDS));
-    } catch (Throwable t) {
-      promise.completeExceptionally(t);
-    }
+        try {
+          if (!policy.isAsync())
+            callable.call();
+          else
+            future.inject((Future) scheduler.schedule(callable, result.getWaitNanos(), TimeUnit.NANOSECONDS));
+        } catch (Throwable t) {
+          promise.completeExceptionally(t);
+        }
+
+        return promise.thenCompose(ss -> postExecuteAsync(ss, scheduler, future));
+      }
 
-    return promise;
+      return postExecuteAsync(result, scheduler, future);
+    });
   }
 }

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

@@ -66,8 +66,8 @@ static class ListenerCounter {
     /** Per listener invocations */
     AtomicInteger invocations = new AtomicInteger();
 
-    /** Records a sync invocation of the {@code listener}. */
-    void sync() {
+    /** Records an invocation of the {@code listener}. */
+    void record() {
       invocations.incrementAndGet();
     }
 
@@ -113,30 +113,30 @@ private <T> FailsafeExecutor<T> registerListeners(RetryPolicy<T> retryPolicy, Ci
         Failsafe.with(retryPolicy, circuitBreaker) :
         Failsafe.with(fallback, retryPolicy, circuitBreaker);
 
-    retryPolicy.onAbort(e -> rpAbort.sync());
-    retryPolicy.onFailedAttempt(e -> rpFailedAttempt.sync());
-    retryPolicy.onRetriesExceeded(e -> rpRetriesExceeded.sync());
-    retryPolicy.onRetry(e -> rpRetry.sync());
-    retryPolicy.onSuccess(e -> rpSuccess.sync());
-    retryPolicy.onFailure(e -> rpFailure.sync());
+    retryPolicy.onAbort(e -> rpAbort.record());
+    retryPolicy.onFailedAttempt(e -> rpFailedAttempt.record());
+    retryPolicy.onRetriesExceeded(e -> rpRetriesExceeded.record());
+    retryPolicy.onRetry(e -> rpRetry.record());
+    retryPolicy.onSuccess(e -> rpSuccess.record());
+    retryPolicy.onFailure(e -> rpFailure.record());
 
-    circuitBreaker.onOpen(() -> cbOpen.sync());
-    circuitBreaker.onHalfOpen(() -> cbHalfOpen.sync());
-    circuitBreaker.onClose(() -> cbClose.sync());
-    circuitBreaker.onSuccess(e -> cbSuccess.sync());
-    circuitBreaker.onFailure(e -> cbFailure.sync());
+    circuitBreaker.onOpen(() -> cbOpen.record());
+    circuitBreaker.onHalfOpen(() -> cbHalfOpen.record());
+    circuitBreaker.onClose(() -> cbClose.record());
+    circuitBreaker.onSuccess(e -> cbSuccess.record());
+    circuitBreaker.onFailure(e -> cbFailure.record());
 
     if (fallback != null) {
-      fallback.onSuccess(e -> fbSuccess.sync());
-      fallback.onFailure(e -> fbFailure.sync());
+      fallback.onSuccess(e -> fbSuccess.record());
+      fallback.onFailure(e -> fbFailure.record());
     }
 
     failsafe.onComplete(e -> {
-      complete.sync();
+      complete.record();
       waiter.resume();
     });
-    failsafe.onSuccess(e -> success.sync());
-    failsafe.onFailure(e -> failure.sync());
+    failsafe.onSuccess(e -> success.record());
+    failsafe.onFailure(e -> failure.record());
 
     return failsafe;
   }
@@ -416,7 +416,7 @@ private void assertForFailingFallback(boolean sync) throws Throwable {
     CircuitBreaker<Object> circuitBreaker = new CircuitBreaker<>().withDelay(Duration.ZERO)
         .handle(NullPointerException.class);
     // And failing Fallback
-    Fallback<Object> fallback = Fallback.ofAsync(() -> true);
+    Fallback<Object> fallback = Fallback.ofAsync(() -> { throw new Exception(); });
     FailsafeExecutor<Object> failsafe = registerListeners(retryPolicy, circuitBreaker, fallback);
 
     // When

src/test/java/net/jodah/failsafe/issues/Issue284Test.java

@@ -0,0 +1,63 @@
+package net.jodah.failsafe.issues;
+
+import net.jodah.failsafe.Failsafe;
+import net.jodah.failsafe.Fallback;
+import net.jodah.failsafe.RetryPolicy;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.testng.Assert.*;
+
+@Test
+public class Issue284Test {
+  AtomicBoolean success;
+  AtomicBoolean failure;
+  AtomicBoolean executed;
+  Fallback<String> fallback;
+  RetryPolicy<String> retryPolicy = new RetryPolicy<String>().handleResult(null)
+    .onSuccess(e -> success.set(true))
+    .onFailure(e -> failure.set(true));
+
+  @BeforeMethod
+  protected void beforeMethod() {
+    success = new AtomicBoolean();
+    failure = new AtomicBoolean();
+    executed = new AtomicBoolean();
+  }
+
+  private Fallback<String> fallbackFor(String result) {
+    return Fallback.of(result).handleResult(null).onSuccess(e -> success.set(true)).onFailure(e -> failure.set(true));
+  }
+
+  public void testFallbackSuccess() {
+    fallback = fallbackFor("hello");
+    String result = Failsafe.with(fallback).get(() -> null);
+
+    assertEquals(result, "hello");
+    assertTrue(success.get(), "Fallback should have been successful");
+  }
+
+  public void testFallbackFailure() {
+    fallback = fallbackFor(null);
+    String result = Failsafe.with(fallback).get(() -> null);
+
+    assertNull(result);
+    assertTrue(failure.get(), "Fallback should have failed");
+  }
+
+  public void testRetryPolicySuccess() {
+    String result = Failsafe.with(retryPolicy).get(() -> !executed.getAndSet(true) ? null : "hello");
+
+    assertEquals(result, "hello");
+    assertTrue(success.get(), "RetryPolicy should have been successful");
+  }
+
+  public void testRetryPolicyFailure() {
+    String result = Failsafe.with(retryPolicy).get(() -> null);
+
+    assertNull(result);
+    assertTrue(failure.get(), "RetryPolicy should have failed");
+  }
+}