async driver request timeout

pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.arangodb</groupId>
     <artifactId>arangodb-java-driver</artifactId>
-    <version>6.11.1</version>
+    <version>6.11.2-SNAPSHOT</version>
     <inceptionYear>2016</inceptionYear>
     <packaging>jar</packaging>
 

src/main/java/com/arangodb/async/internal/utils/CompletableFutureUtils.java

@@ -0,0 +1,96 @@
+package com.arangodb.async.internal.utils;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.BiConsumer;
+
+public class CompletableFutureUtils {
+    /**
+     * Singleton delay scheduler, used only for starting and cancelling tasks.
+     */
+    static final class Delayer {
+        static ScheduledFuture<?> delay(Runnable command, long delay,
+                                        TimeUnit unit) {
+            return delayer.schedule(command, delay, unit);
+        }
+
+        static final class DaemonThreadFactory implements ThreadFactory {
+            public Thread newThread(Runnable r) {
+                Thread t = new Thread(r);
+                t.setDaemon(true);
+                t.setName("CompletableFutureDelayScheduler");
+                return t;
+            }
+        }
+
+        static final ScheduledThreadPoolExecutor delayer;
+
+        static {
+            (delayer = new ScheduledThreadPoolExecutor(
+                    1, new DaemonThreadFactory())).
+                    setRemoveOnCancelPolicy(true);
+        }
+    }
+
+    /**
+     * Action to cancel unneeded timeouts
+     */
+    static final class Canceller implements BiConsumer<Object, Throwable> {
+        final Future<?> f;
+
+        Canceller(Future<?> f) {
+            this.f = f;
+        }
+
+        public void accept(Object ignore, Throwable ex) {
+            if (ex == null && f != null && !f.isDone())
+                f.cancel(false);
+        }
+    }
+
+    /**
+     * Action to completeExceptionally on timeout
+     */
+    static final class Timeout implements Runnable {
+        final CompletableFuture<?> f;
+
+        Timeout(CompletableFuture<?> f) {
+            this.f = f;
+        }
+
+        public void run() {
+            if (f != null && !f.isDone())
+                f.completeExceptionally(new TimeoutException());
+        }
+    }
+
+    /**
+     * Exceptionally completes {@code completableFuture} with a {@link TimeoutException} if not otherwise completed
+     * before the given timeout.
+     *
+     * @param completableFuture
+     *         the original CompletableFuture
+     * @param timeout
+     *         how long to wait before completing exceptionally with a TimeoutException, in units of {@code unit}
+     * @param unit
+     *         a {@code TimeUnit} determining how to interpret the {@code timeout} parameter
+     * @return this CompletableFuture
+     *
+     * @since 9
+     */
+    public static <T> CompletableFuture<T> orTimeout(CompletableFuture<T> completableFuture, long timeout, TimeUnit unit) {
+        if (unit == null)
+            throw new NullPointerException();
+        if (!completableFuture.isDone())
+            completableFuture.whenComplete(new Canceller(Delayer.delay(new Timeout(completableFuture),
+                    timeout, unit)));
+        return completableFuture;
+    }
+
+
+}

src/main/java/com/arangodb/async/internal/velocystream/VstConnectionAsync.java

@@ -20,6 +20,7 @@
 
 package com.arangodb.async.internal.velocystream;
 
+import com.arangodb.async.internal.utils.CompletableFutureUtils;
 import com.arangodb.internal.net.HostDescription;
 import com.arangodb.internal.velocystream.internal.Chunk;
 import com.arangodb.internal.velocystream.internal.Message;
@@ -30,6 +31,7 @@
 import java.util.Collection;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author Mark Vollmary
@@ -54,7 +56,7 @@ public synchronized CompletableFuture<Message> write(final Message message, fina
         });
         messageStore.storeMessage(message.getId(), task);
         super.writeIntern(message, chunks);
-        return future;
+        return CompletableFutureUtils.orTimeout(future, timeout, TimeUnit.MILLISECONDS);
     }
 
     @Override

src/test/resources/arangodb.properties

@@ -2,3 +2,4 @@ arangodb.hosts=172.28.3.1:8529
 arangodb.connections.max=1
 arangodb.acquireHostList=true
 arangodb.password=test
+arangodb.timeout=5000