WebSockets Next: introduce connection-idle-timeout config property

Author: mkouba

extensions/websockets-next/deployment/src/test/java/io/quarkus/websockets/next/test/connection/ConnectionIdleTimeoutTest.java

@@ -0,0 +1,96 @@
+package io.quarkus.websockets.next.test.connection;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import jakarta.inject.Inject;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.quarkus.websockets.next.OnClose;
+import io.quarkus.websockets.next.OnOpen;
+import io.quarkus.websockets.next.OnTextMessage;
+import io.quarkus.websockets.next.WebSocket;
+import io.quarkus.websockets.next.WebSocketClient;
+import io.quarkus.websockets.next.WebSocketClientConnection;
+import io.quarkus.websockets.next.WebSocketConnector;
+import io.quarkus.websockets.next.test.utils.WSClient;
+
+public class ConnectionIdleTimeoutTest {
+
+    @RegisterExtension
+    public static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot(root -> {
+                root.addClasses(ServerEndpoint.class, ClientEndpoint.class, WSClient.class);
+            }).overrideConfigKey("quarkus.websockets-next.client.connection-idle-timeout", "500ms");;
+
+    @TestHTTPResource("/")
+    URI uri;
+
+    @Inject
+    WebSocketConnector<ClientEndpoint> connector;
+
+    @Test
+    public void testTimeout() throws InterruptedException {
+        WebSocketClientConnection conn = connector.baseUri(uri.toString()).connectAndAwait();
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        try {
+            TimeUnit.MILLISECONDS.sleep(500);
+            executor.execute(() -> {
+                try {
+                    conn.sendTextAndAwait("ok");
+                } catch (Throwable ignored) {
+                }
+            });
+        } finally {
+            executor.shutdownNow();
+        }
+        assertTrue(ServerEndpoint.CLOSED.await(5, TimeUnit.SECONDS));
+        assertTrue(ClientEndpoint.CLOSED.await(5, TimeUnit.SECONDS));
+        assertFalse(ServerEndpoint.MESSAGE.get());
+    }
+
+    @WebSocket(path = "/end")
+    public static class ServerEndpoint {
+
+        static final CountDownLatch CLOSED = new CountDownLatch(1);
+        static final AtomicBoolean MESSAGE = new AtomicBoolean();
+
+        @OnTextMessage
+        void onText(String message) {
+            MESSAGE.set(true);
+        }
+
+        @OnClose
+        void close() {
+            CLOSED.countDown();
+        }
+
+    }
+
+    @WebSocketClient(path = "/end")
+    public static class ClientEndpoint {
+
+        static final CountDownLatch CLOSED = new CountDownLatch(1);
+
+        @OnOpen
+        void open() {
+        }
+
+        @OnClose
+        void close(WebSocketClientConnection conn) {
+            CLOSED.countDown();
+        }
+
+    }
+}

extensions/websockets-next/runtime/src/main/java/io/quarkus/websockets/next/runtime/WebSocketConnectorBase.java

@@ -3,6 +3,7 @@
 import java.net.URI;
 import java.net.URLEncoder;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -11,6 +12,7 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -181,7 +183,16 @@ protected WebSocketClientOptions populateClientOptions() {
         if (config.maxFrameSize().isPresent()) {
             clientOptions.setMaxFrameSize(config.maxFrameSize().getAsInt());
         }
-
+        if (config.connectionIdleTimeout().isPresent()) {
+            Duration timeout = config.connectionIdleTimeout().get();
+            if (timeout.toMillis() > Integer.MAX_VALUE) {
+                clientOptions.setIdleTimeoutUnit(TimeUnit.SECONDS);
+                clientOptions.setIdleTimeout((int) timeout.toSeconds());
+            } else {
+                clientOptions.setIdleTimeoutUnit(TimeUnit.MILLISECONDS);
+                clientOptions.setIdleTimeout((int) timeout.toMillis());
+            }
+        }
         Optional<TlsConfiguration> maybeTlsConfiguration = TlsConfiguration.from(tlsConfigurationRegistry,
                 Optional.ofNullable(tlsConfigurationName));
         if (maybeTlsConfiguration.isEmpty()) {

extensions/websockets-next/runtime/src/main/java/io/quarkus/websockets/next/runtime/config/WebSocketsClientRuntimeConfig.java

@@ -47,6 +47,11 @@ public interface WebSocketsClientRuntimeConfig {
      */
     Optional<Duration> autoPingInterval();
 
+    /**
+     * If set then a connection will be closed if no data is received nor sent within the given timeout.
+     */
+    Optional<Duration> connectionIdleTimeout();
+
     /**
      * The strategy used when an error occurs but no error handler can handle the failure.
      * <p>