CircuitBreaker + Predicate + Failure interaction. (#131)

* Failing test with CircuitBreaker + Predicate + Failure interaction.

* Tests are only run if they end in *Test.java!

http://maven.apache.org/surefire/maven-surefire-plugin/examples/inclusion-exclusion.html

* Went ahead and implemented fixes.

* Fill in test coverage for missing branches.

* Remove PR discussion comments. Ready to merge.

* Minor comment fix.

* Add RetryPolicy Predicate exception-handling.

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

@@ -109,7 +109,8 @@ public <T> CircuitBreaker failIf(BiPredicate<T, ? extends Throwable> completionP
 
   /**
    * Specifies that a failure should be recorded if the {@code resultPredicate} matches the result.
-   * 
+   * Predicate is not invoked when the operation fails.
+   *
    * @throws NullPointerException if {@code resultPredicate} is null
    */
   public <T> CircuitBreaker failIf(Predicate<T> resultPredicate) {
@@ -250,8 +251,13 @@ public boolean isClosed() {
    */
   public boolean isFailure(Object result, Throwable failure) {
     for (BiPredicate<Object, Throwable> predicate : failureConditions) {
-      if (predicate.test(result, failure))
-        return true;
+      try {
+        if (predicate.test(result, failure))
+          return true;
+      } catch (Exception t) {
+        // Ignore confused user-supplied predicates.
+        // They should not be allowed to halt execution of the operation.
+      }
     }
 
     // Return true if the failure is not checked by a configured condition

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

@@ -52,14 +52,24 @@ public boolean test(Object t, Throwable u) {
   }
 
   /**
-   * Returns a predicate that evaluates the {@code resultPredicate} against a result.
+   * Returns a predicate that evaluates the {@code resultPredicate} against a result, when present.
+   *
+   * Short-circuts to false without invoking {@code resultPredicate},
+   * when result is not present (i.e. biPredicate.test(null, Throwable)).
    */
   @SuppressWarnings("unchecked")
   static <T> BiPredicate<Object, Throwable> resultPredicateFor(final Predicate<T> resultPredicate) {
     return new BiPredicate<Object, Throwable>() {
       @Override
       public boolean test(Object t, Throwable u) {
-        return ((Predicate<Object>) resultPredicate).test(t);
+        if (u == null) {
+          return ((Predicate<Object>) resultPredicate).test(t);
+        } else {
+          // resultPredicate is only defined over the success type.
+          // It doesn't know how to handle a failure of type Throwable,
+          // so we return false here.
+          return false;
+        }
       }
     };
   }

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

@@ -129,7 +129,8 @@ public <T> RetryPolicy abortIf(BiPredicate<T, ? extends Throwable> completionPre
 
   /**
    * Specifies that retries should be aborted if the {@code resultPredicate} matches the result.
-   * 
+   * Predicate is not invoked when the operation fails.
+   *
    * @throws NullPointerException if {@code resultPredicate} is null
    */
   public <T> RetryPolicy abortIf(Predicate<T> resultPredicate) {
@@ -220,8 +221,13 @@ public boolean allowsRetries() {
    */
   public boolean canAbortFor(Object result, Throwable failure) {
     for (BiPredicate<Object, Throwable> predicate : abortConditions) {
-      if (predicate.test(result, failure))
-        return true;
+      try {
+        if (predicate.test(result, failure))
+          return true;
+      } catch (Exception t) {
+        // Ignore confused user-supplied predicates.
+        // They should not be allowed to halt execution of the operation.
+      }
     }
     return false;
   }
@@ -238,8 +244,13 @@ public boolean canAbortFor(Object result, Throwable failure) {
    */
   public boolean canRetryFor(Object result, Throwable failure) {
     for (BiPredicate<Object, Throwable> predicate : retryConditions) {
-      if (predicate.test(result, failure))
-        return true;
+      try {
+        if (predicate.test(result, failure))
+          return true;
+      } catch (Exception t) {
+        // Ignore confused user-supplied predicates.
+        // They should not be allowed to halt execution of the operation.
+      }
     }
 
     // Retry by default if a failure is not checked by a retry condition
@@ -376,6 +387,7 @@ public <T> RetryPolicy retryIf(BiPredicate<T, ? extends Throwable> completionPre
   /**
    * Specifies that a retry should occur if the {@code resultPredicate} matches the result and the retry policy is not
    * exceeded.
+   * Predicate is not invoked when the operation fails.
    * 
    * @throws NullPointerException if {@code resultPredicate} is null
    */

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

@@ -24,6 +24,7 @@
 import java.util.Arrays;
 import java.util.concurrent.TimeUnit;
 
+import net.jodah.failsafe.function.BiPredicate;
 import org.testng.annotations.Test;
 
 @Test
@@ -45,6 +46,27 @@ public void testIsFailureForResultPredicate() {
     assertFalse(breaker.isFailure(50, null));
   }
 
+  public void testIgnoresThrowingPredicate() {
+    CircuitBreaker breaker = new CircuitBreaker().failIf(new BiPredicate<Integer, Throwable>() {
+      @Override
+      public boolean test(Integer integer, Throwable throwable) {
+        throw new NullPointerException();
+      }
+    });
+    assertFalse(breaker.isFailure(1, null));
+  }
+
+  @Test(expectedExceptions = OutOfMemoryError.class)
+  public void testThrowsFatalErrors() {
+    CircuitBreaker breaker = new CircuitBreaker().failIf(new BiPredicate<String, Throwable>() {
+      @Override
+      public boolean test(String integer, Throwable throwable) {
+        throw new OutOfMemoryError();
+      }
+    });
+    breaker.isFailure("result", null);
+  }
+
   @SuppressWarnings("unchecked")
   public void testIsFailureForFailure() {
     CircuitBreaker breaker = new CircuitBreaker();

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

@@ -0,0 +1,31 @@
+/*
+ * 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.function.BiPredicate;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+@Test
+public class PredicatesTest {
+  public void testResultPredicateOnlyHandlesResults() {
+    BiPredicate<Object, Throwable> resultPredicate = Predicates.resultPredicateFor(result -> true);
+    Assert.assertTrue(resultPredicate.test("result", null));
+    Assert.assertFalse(resultPredicate.test(null, new RuntimeException()));
+  }
+}

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

@@ -0,0 +1,85 @@
+/*
+ * 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.issues;
+
+import net.jodah.concurrentunit.Waiter;
+import net.jodah.failsafe.*;
+import net.jodah.failsafe.function.Predicate;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+import java.util.concurrent.*;
+
+import static org.testng.Assert.assertFalse;
+
+@Test
+public class Issue131Test {
+
+  /**
+   * This predicate is invoked in failure scenarios with an arg of null,
+   * producing a {@link NullPointerException} yielding surpising results.
+   */
+  private static Predicate<String> failIfEqualsIgnoreCaseFoo = new Predicate<String>() {
+    @Override
+    public boolean test(String s) {
+      return s.equalsIgnoreCase("foo"); // produces NPE when invoked in failing scenarios.
+    }
+  };
+
+  /**
+   * Simple synchronous case throwing a {@link NullPointerException}
+   * instead of the expected {@link FailsafeException}.
+   */
+  @Test(expectedExceptions = FailsafeException.class)
+  public void syncShouldThrowTheUnderlyingIOException() throws Throwable {
+    CircuitBreaker circuitBreaker = new CircuitBreaker().failIf(failIfEqualsIgnoreCaseFoo);
+    SyncFailsafe<String> failsafe = Failsafe.<String>with(circuitBreaker);
+
+    // I expect this get() to throw IOException, not NPE.
+    failsafe.get(new Callable<String>() {
+      @Override
+      public String call() throws Exception {
+        throw new IOException("let's blame it on network error");
+      }
+    });
+  }
+
+
+  /**
+   * More alarming async case where the Future is not even completed
+   * since Failsafe does not recover from the {@link NullPointerException} thrown by the predicate.
+   */
+  public void asyncShouldCompleteTheFuture() throws Throwable {
+    CircuitBreaker circuitBreaker = new CircuitBreaker().failIf(failIfEqualsIgnoreCaseFoo);
+    AsyncFailsafe<String> failsafe = Failsafe.<String>with(circuitBreaker).with(Executors.newSingleThreadScheduledExecutor());
+
+    Waiter waiter = new Waiter();
+
+    failsafe
+      .future(new Callable<CompletionStage<String>>() {
+        @Override
+        public CompletionStage<String> call() throws Exception {
+          CompletableFuture<String> future = new CompletableFuture<String>();
+          future.completeExceptionally(new IOException("let's blame it on network error"));
+          return future;
+        }
+      })
+      .whenComplete((s, t) -> waiter.resume()); // Never invoked!
+
+    waiter.await(1000);
+  }
+}