Add test cases involving potentially blocked operations

Author: jhalterman

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

@@ -0,0 +1,102 @@
+package net.jodah.failsafe.functional;
+
+import net.jodah.failsafe.*;
+import org.testng.annotations.Test;
+
+import java.time.Duration;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.testng.Assert.assertFalse;
+
+/**
+ * Tests scenarios against a small threadpool where executions could be temporarily blocked.
+ */
+@Test
+public class BlockedExecutionTest {
+  /**
+   * Asserts that a scheduled execution that is blocked on a threadpool is properly cancelled when a timeout occurs.
+   */
+  @Test
+  public void shouldCancelScheduledExecutionOnTimeout() throws Throwable {
+    ExecutorService executor = Executors.newSingleThreadExecutor();
+    Timeout<Boolean> timeout = Timeout.of(Duration.ofMillis(100));
+    AtomicBoolean supplierCalled = new AtomicBoolean();
+    executor.submit(Testing.uncheck(() -> Thread.sleep(300)));
+
+    Future<Boolean> future = Failsafe.with(timeout).with(executor).getAsync(() -> {
+      supplierCalled.set(true);
+      return false;
+    });
+
+    Asserts.assertThrows(() -> future.get(1000, TimeUnit.MILLISECONDS), ExecutionException.class,
+      TimeoutExceededException.class);
+    Thread.sleep(300);
+    assertFalse(supplierCalled.get());
+  }
+
+  /**
+   * Asserts that a scheduled retry that is blocked on a threadpool is properly cancelled when a timeout occurs.
+   */
+  public void shouldCancelScheduledRetryOnTimeout() {
+    ExecutorService executor = Executors.newSingleThreadExecutor();
+    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(() -> {
+      // Tie up single thread immediately after execution, before the retry is scheduled
+      executor.submit(Testing.uncheck(() -> Thread.sleep(1000)));
+      return false;
+    });
+
+    Asserts.assertThrows(() -> future.get(500, TimeUnit.MILLISECONDS), ExecutionException.class,
+      TimeoutExceededException.class);
+  }
+
+  /**
+   * Asserts that a scheduled fallback that is blocked on a threadpool is properly cancelled when a timeout occurs.
+   */
+  public void shouldCancelScheduledFallbackOnTimeout() {
+    ExecutorService executor = Executors.newSingleThreadExecutor();
+    Timeout<Boolean> timeout = Timeout.of(Duration.ofMillis(100));
+    AtomicBoolean fallbackCalled = new AtomicBoolean();
+    Fallback<Boolean> fallback = Fallback.ofAsync(() -> {
+      fallbackCalled.set(true);
+      return true;
+    }).handleResult(false);
+
+    Future<Boolean> future = Failsafe.with(timeout, 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;
+    });
+
+    Asserts.assertThrows(() -> future.get(500, TimeUnit.MILLISECONDS), ExecutionException.class,
+      TimeoutExceededException.class);
+    assertFalse(fallbackCalled.get());
+  }
+
+  /**
+   * Asserts that a scheduled fallback that is blocked on a threadpool is properly cancelled when the outer future is
+   * cancelled.
+   */
+  public void shouldCancelScheduledFallbackOnCancel() throws Throwable {
+    AtomicBoolean fallbackCalled = new AtomicBoolean();
+    ExecutorService executor = Executors.newSingleThreadExecutor();
+    Fallback<Boolean> fallback = Fallback.ofAsync(() -> {
+      fallbackCalled.set(true);
+      return true;
+    }).handleResult(false);
+
+    Future<Boolean> future = Failsafe.with(fallback).with(executor).getAsync(() -> {
+      executor.submit(Testing.uncheck(() -> Thread.sleep(300)));
+      return false;
+    });
+
+    Thread.sleep(100);
+    future.cancel(false);
+    Asserts.assertThrows(future::get, CancellationException.class);
+    Thread.sleep(300);
+    assertFalse(fallbackCalled.get());
+  }
+}

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

@@ -0,0 +1,42 @@
+package net.jodah.failsafe.issues;
+
+import net.jodah.failsafe.Failsafe;
+import net.jodah.failsafe.RetryPolicy;
+import net.jodah.failsafe.Timeout;
+import net.jodah.failsafe.function.ContextualRunnable;
+import org.testng.annotations.Test;
+
+import java.time.Duration;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.function.Function;
+
+@Test
+public class Issue260Test {
+  public void test() throws Throwable {
+    ExecutorService executor = Executors.newSingleThreadExecutor();
+    Timeout<Object> timeout = Timeout.of(Duration.ofMillis(300))
+      .onFailure(e -> System.out.println("Interrupted"))
+      .withInterrupt(true);
+    RetryPolicy<Object> rp = new RetryPolicy<>().onRetry(e -> System.out.println("Retrying"))
+      .onSuccess(e -> System.out.println("Success"));
+
+    Function<Integer, ContextualRunnable> task = (taskId) -> ctx -> {
+      System.out.println("Starting execution of task " + taskId);
+      try {
+        Thread.sleep(200);
+      } catch (InterruptedException e) {
+        System.out.println("Interrupted task " + taskId);
+        throw e;
+      }
+    };
+
+    Future<?> f1 = Failsafe.with(rp, timeout).with(executor).runAsync(task.apply(1));
+    Future<?> f2 = Failsafe.with(rp, timeout).with(executor).runAsync(task.apply(2));
+    Future<?> f3 = Failsafe.with(rp, timeout).with(executor).runAsync(task.apply(3));
+    f1.get();
+    f2.get();
+    f3.get();
+  }
+}