More stuff

Author: krka

README.md

@@ -353,4 +353,3 @@ Otherwise you may be bottlenecking on a static single-threaded `ScheduledExecuto
 # TODO:
 workarounds for timeout-issue
 Explain minimalCompletionStage and delayedExecutor
-Rename OnTimeoutTest -> CompleteOnTimeoutTest

pom.xml

@@ -20,6 +20,12 @@
             <version>4.12</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>28.1-jre</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

src/test/java/se/krka/futures/OnTimeoutTest.java

@@ -1,47 +1,60 @@
 package se.krka.futures;
 
+import com.google.common.util.concurrent.MoreExecutors;
 import org.junit.Test;
 
 import java.io.Closeable;
-import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
 
 public class OnTimeoutTest {
   @Test
   public void testTimeoutNotBlocking() throws Exception {
+    waitForNotBlocking();
     assertTrue(measureTimeout() < 200);
   }
 
   @Test
   public void testTimeoutNotBlockingTwice() throws Exception {
+    waitForNotBlocking();
     assertTrue(measureTimeout() < 200);
     assertTrue(measureTimeout() < 200);
   }
 
   @Test
   public void testTimeoutBlocking() throws Exception {
+    waitForNotBlocking();
     try (Killer killer = new Killer(killTimeout())) {
       expectFailure();
     }
   }
 
   @Test
   public void testTimeoutBlocking2() throws Exception {
+    waitForNotBlocking();
     try (Killer killer = new Killer(killTimeout2())) {
       expectFailure();
     }
   }
 
+  @Test
+  public void testTimeoutBlockingWithDelayedExecutor() throws Exception {
+    waitForNotBlocking();
+    try (Killer killer = new Killer(killTimeout3())) {
+      expectFailure();
+    }
+  }
+
   private void expectFailure() throws Exception {
     long time = measureTimeout();
-    assertTrue("time was " + time, time >= 1000);
+    assertTrue("time was " + time, time >= 900);
   }
 
   private long measureTimeout() throws InterruptedException, TimeoutException {
@@ -59,36 +72,63 @@ private long measureTimeout() throws InterruptedException, TimeoutException {
   }
 
   private CompletableFuture<String> killTimeout() {
-    return new CompletableFuture<String>()
-            .orTimeout(1, TimeUnit.MILLISECONDS)
-            .handle((s, t) -> {
-              try {
-                sleepOnThread();
-                return "";
-              } catch (InterruptedException e) {
-                throw new RuntimeException(e);
-              }
-            })
-            .exceptionally(Throwable::getMessage);
+    waitForNotBlocking();
+    CompletableFuture<String> future = new CompletableFuture<String>()
+            .orTimeout(10, TimeUnit.MILLISECONDS)
+            .exceptionally(Throwable::getMessage)
+            .thenApply(s -> sleepOnThread(s, 1000));
+    waitForBlocking();
+    return future;
   }
 
   private CompletableFuture<String> killTimeout2() {
-    return new CompletableFuture<String>()
-            .completeOnTimeout("", 1, TimeUnit.MILLISECONDS)
-            .thenApply(s -> {
-              try {
-                sleepOnThread();
-                return s;
-              } catch (InterruptedException e) {
-                return s;
-              }
-            });
+    CompletableFuture<String> future = new CompletableFuture<String>()
+            .completeOnTimeout("", 10, TimeUnit.MILLISECONDS)
+            .thenApply(s -> sleepOnThread(s, 1000));
+    waitForBlocking();
+    return future;
   }
 
-  private void sleepOnThread() throws InterruptedException {
-    int millis = 1000;
-    System.out.println("Sleeping on " + Util.currThread() + " for " + millis + " ms");
-    Thread.sleep(millis);
+  private CompletableFuture<String> killTimeout3() {
+    Executor executor = CompletableFuture.delayedExecutor(1, TimeUnit.MILLISECONDS, MoreExecutors.directExecutor());
+    CompletableFuture<String> future = new CompletableFuture<>();
+    executor.execute(() -> {
+      sleepOnThread(null, 1000);
+      future.complete("value");
+    });
+    waitForBlocking();
+    return future;
+  }
+
+  private static final AtomicBoolean BLOCKING_COMPLETABLE_FUTURE_DELAY_SCHEDULER = new AtomicBoolean(false);
+
+  private <T> T sleepOnThread(T value, int millis) {
+    if (Util.currThread().equals("CompletableFutureDelayScheduler")) {
+      BLOCKING_COMPLETABLE_FUTURE_DELAY_SCHEDULER.set(true);
+    }
+    try {
+      System.out.println("Sleeping on " + Util.currThread() + " for " + millis + " ms");
+      Thread.sleep(millis);
+    } catch (InterruptedException e) {
+      e.printStackTrace();
+    } finally {
+      if (Util.currThread().equals("CompletableFutureDelayScheduler")) {
+        BLOCKING_COMPLETABLE_FUTURE_DELAY_SCHEDULER.set(false);
+      }
+    }
+    return value;
+  }
+
+  private void waitForBlocking() {
+    while (!BLOCKING_COMPLETABLE_FUTURE_DELAY_SCHEDULER.get()) {
+      Thread.yield();
+    }
+  }
+
+  private void waitForNotBlocking() {
+    while (BLOCKING_COMPLETABLE_FUTURE_DELAY_SCHEDULER.get()) {
+      Thread.yield();
+    }
   }
 
   private static class Killer implements Closeable {