Change TimeoutExecutor to use internal scheduler

Changes the TimeoutExecutor to use the internal scheduler, so that when a user-provided ScheduledExecutorService is shutdown, Timeout tasks will still fire.

Author: jhalterman

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

@@ -64,7 +64,7 @@ protected Supplier<ExecutionResult> supply(Supplier<ExecutionResult> supplier, S
 
       try {
         // Schedule timeout check
-        timeoutFuture = scheduler.schedule(() -> {
+        timeoutFuture = Scheduler.DEFAULT.schedule(() -> {
           // Guard against race with execution completion
           if (result.compareAndSet(null, ExecutionResult.failure(new TimeoutExceededException(policy)))) {
             // Cancel and interrupt
@@ -114,7 +114,7 @@ protected Supplier<CompletableFuture<ExecutionResult>> supplyAsync(
           if (!future.isDone()) {
             try {
               // Schedule timeout check
-              timeoutFuture.set(scheduler.schedule(() -> {
+              timeoutFuture.set(Scheduler.DEFAULT.schedule(() -> {
                 // Guard against race with execution completion
                 if (executionResult.compareAndSet(null,
                   ExecutionResult.failure(new TimeoutExceededException(policy)))) {

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

@@ -539,45 +539,6 @@ public void shouldSkipExecutionWhenCircuitOpen() {
     assertThrows(future::get, ExecutionException.class, CircuitBreakerOpenException.class);
   }
 
-  /**
-   * Asserts that Failsafe handles an initial scheduling failure.
-   */
-  public void shouldHandleInitialSchedulingFailure() {
-    // Given
-    ScheduledExecutorService executor = Executors.newScheduledThreadPool(0);
-    executor.shutdownNow();
-
-    // When
-    Future future = Failsafe.with(Fallback.of(false), new RetryPolicy<>(), new CircuitBreaker<>())
-      .with(executor)
-      .runAsync(() -> waiter.fail("Should not execute supplier since executor has been shutdown"));
-
-    assertThrows(future::get, ExecutionException.class, RejectedExecutionException.class);
-  }
-
-  /**
-   * Asserts that Failsafe handles a rpRetry scheduling failure.
-   */
-  public void shouldHandleRejectedRetryExecution() throws Throwable {
-    // Given
-    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
-    AtomicInteger counter = new AtomicInteger();
-
-    // When
-    Future future = Failsafe.with(new RetryPolicy<>().handleResult(null).handle(Exception.class))
-      .with(executor)
-      .getAsync(() -> {
-        counter.incrementAndGet();
-        Thread.sleep(200);
-        return null;
-      });
-
-    Thread.sleep(150);
-    executor.shutdownNow();
-    assertThrows(future::get, ExecutionException.class, RejectedExecutionException.class);
-    assertEquals(counter.get(), 1, "Supplier should have been executed before executor was shutdown");
-  }
-
   private void assertInterruptedExceptionOnCancel(FailsafeExecutor<Boolean> failsafe) throws Throwable {
     CompletableFuture<Void> future = failsafe.runAsync(() -> {
       try {

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

@@ -0,0 +1,132 @@
+package net.jodah.failsafe.functional;
+
+import net.jodah.concurrentunit.Waiter;
+import net.jodah.failsafe.*;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.time.Duration;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static net.jodah.failsafe.Asserts.assertThrows;
+import static net.jodah.failsafe.Testing.runAsync;
+import static org.testng.Assert.assertEquals;
+
+/**
+ * Tests the handling of an executor service that is shutdown.
+ */
+@Test
+public class ShutdownExecutorTest {
+  Waiter waiter;
+
+  @BeforeMethod
+  protected void beforeMethod() {
+    waiter = new Waiter();
+  }
+
+  /**
+   * Asserts that Failsafe handles an initial scheduling failure due to an executor being shutdown.
+   */
+  public void shouldHandleInitialSchedulingFailure() {
+    // Given
+    ScheduledExecutorService executor = Executors.newScheduledThreadPool(0);
+    executor.shutdownNow();
+
+    // When
+    Future future = Failsafe.with(Fallback.of(false), new RetryPolicy<>(), new CircuitBreaker<>())
+      .with(executor)
+      .runAsync(() -> waiter.fail("Should not execute supplier since executor has been shutdown"));
+
+    assertThrows(future::get, ExecutionException.class, RejectedExecutionException.class);
+  }
+
+  /**
+   * Asserts that an ExecutorService shutdown() will leave current tasks running while preventing new tasks.
+   */
+  public void shouldHandleShutdown() throws Throwable {
+    // Given
+    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+    AtomicInteger counter = new AtomicInteger();
+
+    // When
+    Future future = Failsafe.with(new RetryPolicy<>()).with(executor).getAsync(() -> {
+      Thread.sleep(200);
+      counter.incrementAndGet();
+      return "success";
+    });
+
+    Thread.sleep(100);
+    executor.shutdown();
+    assertEquals("success", future.get());
+    assertEquals(counter.get(), 1, "Supplier should have completed execution before executor was shutdown");
+
+    future = Failsafe.with(new RetryPolicy<>()).with(executor).getAsync(() -> "test");
+    assertThrows(future::get, ExecutionException.class, RejectedExecutionException.class);
+  }
+
+  /**
+   * Asserts that an ExecutorService shutdown() will interrupt current tasks running and prevent new tasks.
+   */
+  public void shouldHandleShutdownNow() throws Throwable {
+    // Given
+    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+    AtomicInteger counter = new AtomicInteger();
+
+    // When
+    Future future = Failsafe.with(new RetryPolicy<>()).with(executor).runAsync(() -> {
+      Thread.sleep(200);
+      counter.incrementAndGet();
+    });
+
+    Thread.sleep(100);
+    executor.shutdownNow();
+    assertThrows(future::get, ExecutionException.class, RejectedExecutionException.class);
+    assertEquals(counter.get(), 0, "Supplier should have been interrupted after executor shutdownNow");
+  }
+
+  /**
+   * Asserts that an ExecutorService shutdown() will not prevent internally scheduled Timeout tasks from cancelling a
+   * sync execution.
+   */
+  public void testShutdownDoesNotPreventTimeoutSync() throws Throwable {
+    // Given
+    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+    Timeout<Object> timeout = Timeout.of(Duration.ofMillis(200)).withInterrupt(true);
+    AtomicInteger counter = new AtomicInteger();
+
+    // When / then
+    assertThrows(() -> Failsafe.with(timeout).with(executor).run(() -> {
+      Thread.sleep(500);
+      counter.incrementAndGet();
+    }), TimeoutExceededException.class);
+    runAsync(() -> {
+      Thread.sleep(100);
+      executor.shutdown();
+    });
+    assertEquals(counter.get(), 0, "Supplier should have been interrupted after Timeout");
+  }
+
+  /**
+   * Asserts that an ExecutorService shutdown() will not prevent internally scheduled Timeout tasks from cancelling an
+   * async execution.
+   */
+  public void testShutdownDoesNotPreventTimeoutAsync() throws Throwable {
+    // Given
+    ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+    Timeout<Object> timeout = Timeout.of(Duration.ofMillis(200)).withInterrupt(true);
+    AtomicInteger counter = new AtomicInteger();
+
+    // When
+    Future future = Failsafe.with(timeout).with(executor).runAsync(() -> {
+      Thread.sleep(500);
+      counter.incrementAndGet();
+    });
+    Thread.sleep(100);
+    executor.shutdown();
+
+    // Then
+    assertThrows(future::get, ExecutionException.class, TimeoutExceededException.class);
+    assertEquals(counter.get(), 0, "Supplier should have been interrupted after Timeout");
+  }
+}