Add support for explicit policy composition.

Fixes #254

Author: jhalterman

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

@@ -38,7 +38,13 @@ public class Failsafe {
    *   Failsafe.with(fallback, retryPolicy, circuitBreaker).get(supplier);
    * </pre>
    * </p>
-   * This results in the following internal composition when executing a {@code runnable} or {@code supplier} and
+   * <p>
+   * This is equivalent to composition using the the {@link FailsafeExecutor#compose(Policy) compose} method:
+   * <pre>
+   *   Failsafe.with(fallback).compose(retryPolicy).compose(circuitBreaker).get(supplier);
+   * </pre>
+   * </p>
+   * These result in the following internal composition when executing a {@code runnable} or {@code supplier} and
    * handling its result:
    * <p>
    * <pre>

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

@@ -20,6 +20,7 @@
 import net.jodah.failsafe.internal.util.Assert;
 import net.jodah.failsafe.spi.*;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.*;
 import java.util.function.Function;
@@ -53,6 +54,44 @@ public class FailsafeExecutor<R> extends PolicyListeners<FailsafeExecutor<R>, R>
     this.policies = policies;
   }
 
+  /**
+   * Returns the currently configured policies.
+   *
+   * @see #compose(Policy)
+   */
+  public List<? extends Policy<R>> getPolicies() {
+    return policies;
+  }
+
+  /**
+   * Returns a new {@code FailsafeExecutor} that composes the currently configured policies around the given {@code
+   * innerPolicy}. For example, consider:
+   * <p>
+   * <pre>
+   *   Failsafe.with(fallback).compose(retryPolicy).compose(circuitBreaker);
+   * </pre>
+   * </p>
+   * This results in the following internal composition when executing a {@code runnable} or {@code supplier} and
+   * handling its result:
+   * <p>
+   * <pre>
+   *   Fallback(RetryPolicy(CircuitBreaker(Supplier)))
+   * </pre>
+   * </p>
+   * This means the {@code CircuitBreaker} is first to evaluate the {@code Supplier}'s result, then the {@code
+   * RetryPolicy}, then the {@code Fallback}. Each policy makes its own determination as to whether the result
+   * represents a failure. This allows different policies to be used for handling different types of failures.
+   *
+   * @throws NullPointerException if {@code innerPolicy} is null
+   * @see #getPolicies()
+   */
+  public <P extends Policy<R>> FailsafeExecutor<R> compose(P innerPolicy) {
+    Assert.notNull(innerPolicy, "innerPolicy");
+    List<Policy<R>> composed = new ArrayList<>(policies);
+    composed.add(innerPolicy);
+    return new FailsafeExecutor<>(composed);
+  }
+
   /**
    * Executes the {@code supplier} until a successful result is returned or the configured policies are exceeded.
    *

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

@@ -117,4 +117,9 @@ public static <R> Timeout<R> of(Duration timeout) {
   public PolicyExecutor<R, ? extends Policy<R>> toExecutor(int policyIndex) {
     return new TimeoutExecutor<>(this, policyIndex, policyHandlers);
   }
+
+  @Override
+  public String toString() {
+    return "Timeout[timeout=" + timeout + ", interruptable=" + interruptable + ']';
+  }
 }

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

@@ -29,7 +29,6 @@
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
 import static net.jodah.failsafe.Asserts.assertThrows;
@@ -293,51 +292,6 @@ public void shouldCompleteFutureExternally() throws Throwable {
     waiter.await(1000);
   }
 
-  /**
-   * Tests a scenario where three timeouts should cause all delegates to be cancelled with interrupts.
-   */
-  public void shouldCancelNestedTimeoutsWithInterrupt() throws Throwable {
-    // Given
-    RetryPolicy<Boolean> rp = new RetryPolicy<Boolean>().onRetry(e -> System.out.println("Retrying"));
-    Timeout<Boolean> timeout1 = Timeout.of(Duration.ofMillis(1000));
-    Timeout<Boolean> timeout2 = Timeout.<Boolean>of(Duration.ofMillis(200)).withInterrupt(true);
-    AtomicReference<FailsafeFuture<Boolean>> futureRef = new AtomicReference<>();
-    CountDownLatch futureLatch = new CountDownLatch(1);
-
-    // When
-    FailsafeFuture<Boolean> future = (FailsafeFuture<Boolean>) Failsafe.with(rp, timeout2, timeout1).onComplete(e -> {
-      waiter.assertNull(e.getResult());
-      waiter.assertTrue(e.getFailure() instanceof TimeoutExceededException);
-      waiter.resume();
-    }).getAsync(ctx -> {
-      // Wait for futureRef to be set
-      futureLatch.await();
-      waiter.assertTrue(ctx.getLastFailure() == null || ctx.getLastFailure() instanceof TimeoutExceededException);
-
-      try {
-        // Assert not cancelled
-        waiter.assertFalse(ctx.isCancelled());
-        // waiter.assertFalse(futureRef.get().cancelFunctions.isEmpty());
-        Thread.sleep(1000);
-      } catch (InterruptedException e) {
-        // Assert cancelled
-        waiter.assertTrue(ctx.isCancelled());
-        waiter.resume();
-        throw e;
-      }
-      waiter.fail("Expected interruption");
-      return false;
-    });
-    futureRef.set(future);
-    futureLatch.countDown();
-
-    // Then
-    waiter.await(1000, 4);
-    assertFalse(future.isCancelled());
-    assertTrue(future.isDone());
-    assertThrows(future::get, ExecutionException.class, TimeoutExceededException.class);
-  }
-
   private void assertCancel(Function<FailsafeExecutor<?>, Future<?>> executorCallable, Policy<?> policy)
     throws Throwable {
     // Given

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

@@ -43,7 +43,7 @@ public void shouldCancelScheduledRetryOnTimeout() {
     Timeout<Boolean> timeout = Timeout.of(Duration.ofMillis(100));
     RetryPolicy<Boolean> rp = new RetryPolicy<Boolean>().withDelay(Duration.ofMillis(1000)).handleResult(false);
 
-    Future<Boolean> future = Failsafe.with(timeout, rp).with(executor).getAsync(() -> {
+    Future<Boolean> future = Failsafe.with(timeout).compose(rp).with(executor).getAsync(() -> {
       // Tie up single thread immediately after execution, before the retry is scheduled
       executor.submit(Testing.uncheck(() -> Thread.sleep(1000)));
       return false;
@@ -65,7 +65,7 @@ public void shouldCancelScheduledFallbackOnTimeout() {
       return true;
     }).handleResult(false);
 
-    Future<Boolean> future = Failsafe.with(timeout, fallback).with(executor).getAsync(() -> {
+    Future<Boolean> future = Failsafe.with(timeout).compose(fallback).with(executor).getAsync(() -> {
       // Tie up single thread immediately after execution, before the fallback is scheduled
       executor.submit(Testing.uncheck(() -> Thread.sleep(1000)));
       return false;

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

@@ -1,12 +1,15 @@
 package net.jodah.failsafe.functional;
 
+import net.jodah.concurrentunit.Waiter;
 import net.jodah.failsafe.*;
 import org.testng.annotations.Test;
 
 import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.*;
 
 /**
  * Tests nested timeout scenarios.
@@ -84,4 +87,49 @@ public void testFallbackRetryPolicyTimeoutTimeout() {
     innerTimeout.withInterrupt(true);
     test.run();
   }
+
+  /**
+   * Tests a scenario where three timeouts should cause all delegates to be cancelled with interrupts.
+   */
+  // TODO consider removing this test in favor of the ones above
+  public void shouldCancelNestedTimeoutsWithInterrupt() throws Throwable {
+    // Given
+    RetryPolicy<Boolean> rp = new RetryPolicy<Boolean>().onRetry(e -> System.out.println("Retrying"));
+    Timeout<Boolean> timeout1 = Timeout.of(Duration.ofMillis(1000));
+    Timeout<Boolean> timeout2 = Timeout.<Boolean>of(Duration.ofMillis(200)).withInterrupt(true);
+    CountDownLatch futureLatch = new CountDownLatch(1);
+    Waiter waiter = new Waiter();
+
+    // When
+    Future<Boolean> future = Failsafe.with(rp).compose(timeout2).compose(timeout1).onComplete(e -> {
+      waiter.assertNull(e.getResult());
+      waiter.assertTrue(e.getFailure() instanceof TimeoutExceededException);
+      waiter.resume();
+    }).getAsync(ctx -> {
+      // Wait for futureRef to be set
+      futureLatch.await();
+      waiter.assertTrue(ctx.getLastFailure() == null || ctx.getLastFailure() instanceof TimeoutExceededException);
+
+      try {
+        // Assert not cancelled
+        waiter.assertFalse(ctx.isCancelled());
+        // waiter.assertFalse(futureRef.get().cancelFunctions.isEmpty());
+        Thread.sleep(1000);
+      } catch (InterruptedException e) {
+        // Assert cancelled
+        waiter.assertTrue(ctx.isCancelled());
+        waiter.resume();
+        throw e;
+      }
+      waiter.fail("Expected interruption");
+      return false;
+    });
+    futureLatch.countDown();
+
+    // Then
+    waiter.await(1000, 4);
+    assertFalse(future.isCancelled());
+    assertTrue(future.isDone());
+    assertThrows(future::get, ExecutionException.class, TimeoutExceededException.class);
+  }
 }

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

@@ -36,7 +36,7 @@ public void testFallbackRetryPolicyCircuitBreaker() {
 
     testRunSuccess(() -> {
       resetBreaker(cb);
-    }, Failsafe.with(fb, rp, cb), ctx -> {
+    }, Failsafe.with(fb).compose(rp).compose(cb), ctx -> {
       throw new IllegalStateException();
     }, e -> {
       assertEquals(cb.getFailureCount(), 3);
@@ -54,7 +54,7 @@ public void testCircuitBreakerRetryPolicy() {
 
     testRunFailure(() -> {
       resetBreaker(cb);
-    }, Failsafe.with(cb, rp), ctx -> {
+    }, Failsafe.with(cb).compose(rp), ctx -> {
       throw new IllegalStateException();
     }, e -> {
       assertEquals(e.getAttemptCount(), 3);
@@ -71,7 +71,7 @@ public void testFallbackRetryPolicy() {
     RetryPolicy<Object> rp = new RetryPolicy<>().withMaxRetries(2);
     Fallback<Object> fb = Fallback.of("test");
 
-    testRunSuccess(Failsafe.with(fb, rp), ctx -> {
+    testRunSuccess(Failsafe.with(fb).compose(rp), ctx -> {
       throw new IllegalStateException();
     }, e -> {
       assertEquals(e.getAttemptCount(), 3);
@@ -85,7 +85,7 @@ public void testRetryPolicyFallback() {
     RetryPolicy<Object> rp = new RetryPolicy<>().withMaxRetries(2);
     Fallback<Object> fb = Fallback.of("test");
 
-    testRunSuccess(Failsafe.with(rp, fb), ctx -> {
+    testRunSuccess(Failsafe.with(rp).compose(fb), ctx -> {
       throw new IllegalStateException();
     }, e -> {
       assertEquals(e.getAttemptCount(), 1);

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

@@ -163,7 +163,7 @@ public void testRetryThenTimeoutWithPendingRetry() {
       executionCounter.set(0);
       timeoutStats.reset();
       rpStats.reset();
-    }, Failsafe.with(timeout, retryPolicy), ctx -> {
+    }, Failsafe.with(timeout).compose(retryPolicy), ctx -> {
       System.out.println("Executing");
       executionCounter.incrementAndGet();
       throw new Exception();
@@ -198,7 +198,7 @@ public void testTimeoutThenFallbackWithBlockedSupplier() {
     Runnable test = () -> testRunFailure(() -> {
       timeoutStats.reset();
       fbStats.reset();
-    }, Failsafe.with(fallback, timeout), ctx -> {
+    }, Failsafe.with(fallback).compose(timeout), ctx -> {
       System.out.println("Executing");
       Thread.sleep(100);
       throw new Exception();
@@ -233,7 +233,7 @@ public void testTimeoutThenFallback() {
     Runnable test = () -> testRunFailure(() -> {
       timeoutStats.reset();
       fbStats.reset();
-    }, Failsafe.with(fallback, timeout), ctx -> {
+    }, Failsafe.with(fallback).compose(timeout), ctx -> {
       System.out.println("Executing");
       throw new Exception();
     }, e -> {
@@ -266,7 +266,7 @@ public void testFallbackThenTimeoutWithBlockedSupplier() {
     Runnable test = () -> testRunFailure(() -> {
       timeoutStats.reset();
       fbStats.reset();
-    }, Failsafe.with(timeout, fallback), ctx -> {
+    }, Failsafe.with(timeout).compose(fallback), ctx -> {
       System.out.println("Executing");
       Thread.sleep(100);
       throw new Exception();
@@ -301,7 +301,7 @@ public void testFallbackThenTimeoutWithBlockedFallback() {
     Runnable test = () -> testRunFailure(() -> {
       timeoutStats.reset();
       fbStats.reset();
-    }, Failsafe.with(timeout, fallback), ctx -> {
+    }, Failsafe.with(timeout).compose(fallback), ctx -> {
       System.out.println("Executing");
       throw new Exception();
     }, e -> {