WebSockets Next: enable traffic logging for debugging purposes

- resolves quarkusio#37947

docs/src/main/asciidoc/websockets-next-reference.adoc

@@ -765,6 +765,29 @@ public class ExampleHttpUpgradeCheck implements HttpUpgradeCheck {
 
 TIP: You can choose WebSocket endpoints to which the `HttpUpgradeCheck` is applied with the `HttpUpgradeCheck#appliesTo` method.
 
+[[traffic-logging]]
+== Traffic logging 
+
+Quarkus can log the messages sent and received for debugging purposes. 
+To enable traffic logging, set the `quarkus.websockets-next.server.traffic-logging.enabled` configuration property to `true`.
+The payload of text messages is logged as well.
+However, the number of logged characters is limited.
+The default limit is 100, but you can change this limit with the `quarkus.websockets-next.server.traffic-logging.text-payload-limit` configuration property.
+
+TIP: The messages are only logged if the `DEBUG` level is enabled for the logger `io.quarkus.websockets.next.traffic`.
+
+.Example configuration
+[source, properties]
+----
+quarkus.websockets-next.server.traffic-logging.enabled=true <1>
+quarkus.websockets-next.server.traffic-logging.text-payload-limit=50 <2>
+
+quarkus.log.category."io.quarkus.websockets.next.traffic".level=DEBUG <3>
+----
+<1> Enables traffic logging.
+<2> Set the number of characters of a text message payload which will be logged.
+<3> Enable `DEBUG` level is for the logger `io.quarkus.websockets.next.traffic`.
+
 [[websocket-next-configuration-reference]]
 == Configuration reference
 

extensions/websockets-next/deployment/src/test/java/io/quarkus/websockets/next/test/traffic/BasicConnectorTrafficLoggerTest.java

@@ -0,0 +1,57 @@
+package io.quarkus.websockets.next.test.traffic;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+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.websockets.next.BasicWebSocketConnector;
+import io.quarkus.websockets.next.WebSocketClientConnection;
+import io.vertx.core.Context;
+
+public class BasicConnectorTrafficLoggerTest extends TrafficLoggerTest {
+
+    @RegisterExtension
+    public static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot(root -> {
+                root.addClasses(Endpoint.class);
+                TrafficLoggerTest.addApplicationProperties(root, false);
+            })
+            .setLogRecordPredicate(TrafficLoggerTest::isTrafficLogRecord)
+            .assertLogRecords(logRecordsConsumer(true));
+
+    @Inject
+    BasicWebSocketConnector connector;
+
+    @Test
+    public void testTrafficLogger() throws InterruptedException {
+        List<String> messages = new CopyOnWriteArrayList<>();
+        CountDownLatch closedLatch = new CountDownLatch(1);
+        CountDownLatch messageLatch = new CountDownLatch(1);
+        WebSocketClientConnection conn = connector
+                .baseUri(endUri)
+                .path("end")
+                .onTextMessage((c, m) -> {
+                    assertTrue(Context.isOnWorkerThread());
+                    messages.add(m);
+                    messageLatch.countDown();
+                })
+                .onClose((c, s) -> closedLatch.countDown())
+                .connectAndAwait();
+        conn.sendTextAndAwait("dummy");
+        assertTrue(messageLatch.await(5, TimeUnit.SECONDS));
+        assertEquals("ok", messages.get(0));
+        conn.closeAndAwait();
+        assertTrue(closedLatch.await(5, TimeUnit.SECONDS));
+    }
+
+}

extensions/websockets-next/deployment/src/test/java/io/quarkus/websockets/next/test/traffic/ClientEndpointTrafficLoggerTest.java

@@ -0,0 +1,72 @@
+package io.quarkus.websockets.next.test.traffic;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+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.websockets.next.OnClose;
+import io.quarkus.websockets.next.OnTextMessage;
+import io.quarkus.websockets.next.WebSocketClient;
+import io.quarkus.websockets.next.WebSocketClientConnection;
+import io.quarkus.websockets.next.WebSocketConnector;
+
+public class ClientEndpointTrafficLoggerTest extends TrafficLoggerTest {
+
+    @RegisterExtension
+    public static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot(root -> {
+                root.addClasses(Endpoint.class, Client.class);
+                TrafficLoggerTest.addApplicationProperties(root, false);
+            })
+            .setLogRecordPredicate(
+                    TrafficLoggerTest::isTrafficLogRecord)
+            .assertLogRecords(logRecordsConsumer(true));
+
+    @Inject
+    WebSocketConnector<Client> connector;
+
+    @Test
+    public void testTrafficLogger() throws InterruptedException {
+        WebSocketClientConnection conn = connector
+                .baseUri(endUri)
+                .connectAndAwait();
+        assertTrue(Client.MESSAGE_LATCH.await(5, TimeUnit.SECONDS));
+        assertEquals("ok", Client.MESSAGES.get(0));
+        conn.closeAndAwait();
+        assertTrue(Client.CLOSED_LATCH.await(5, TimeUnit.SECONDS));
+        assertTrue(Endpoint.CLOSED_LATCH.await(5, TimeUnit.SECONDS));
+    }
+
+    @WebSocketClient(path = "/end")
+    public static class Client {
+
+        static final List<String> MESSAGES = new CopyOnWriteArrayList<>();
+
+        static final CountDownLatch MESSAGE_LATCH = new CountDownLatch(1);
+
+        static final CountDownLatch CLOSED_LATCH = new CountDownLatch(1);
+
+        @OnTextMessage
+        void onMessage(String message) {
+            MESSAGES.add(message);
+            MESSAGE_LATCH.countDown();
+        }
+
+        @OnClose
+        void onClose() {
+            CLOSED_LATCH.countDown();
+        }
+
+    }
+
+}

extensions/websockets-next/deployment/src/test/java/io/quarkus/websockets/next/test/traffic/ServerTrafficLoggerTest.java

@@ -0,0 +1,42 @@
+package io.quarkus.websockets.next.test.traffic;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+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.websockets.next.test.utils.WSClient;
+import io.vertx.core.Vertx;
+
+public class ServerTrafficLoggerTest extends TrafficLoggerTest {
+
+    @RegisterExtension
+    public static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot(root -> {
+                root.addClasses(Endpoint.class, WSClient.class);
+                TrafficLoggerTest.addApplicationProperties(root, true);
+            })
+            .setLogRecordPredicate(TrafficLoggerTest::isTrafficLogRecord)
+            .assertLogRecords(logRecordsConsumer(false));
+
+    @Inject
+    Vertx vertx;
+
+    @Test
+    public void testTrafficLogger() throws InterruptedException, ExecutionException {
+        try (WSClient client = new WSClient(vertx)) {
+            client.connect(WSClient.toWS(endUri, "end"));
+            client.waitForMessages(1);
+            assertEquals("ok", client.getMessages().get(0).toString());
+        }
+        assertTrue(Endpoint.CLOSED_LATCH.await(5, TimeUnit.SECONDS));
+    }
+
+}

extensions/websockets-next/deployment/src/test/java/io/quarkus/websockets/next/test/traffic/TrafficLoggerTest.java

@@ -0,0 +1,66 @@
+package io.quarkus.websockets.next.test.traffic;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.net.URI;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.function.Consumer;
+import java.util.logging.Level;
+import java.util.logging.LogRecord;
+
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.quarkus.websockets.next.OnClose;
+import io.quarkus.websockets.next.OnOpen;
+import io.quarkus.websockets.next.WebSocket;
+
+public abstract class TrafficLoggerTest {
+
+    @TestHTTPResource("/")
+    URI endUri;
+
+    @WebSocket(path = "/end")
+    public static class Endpoint {
+
+        static final CountDownLatch CLOSED_LATCH = new CountDownLatch(1);
+
+        @OnOpen
+        public String open() {
+            return "ok";
+        }
+
+        @OnClose
+        public void close() {
+            CLOSED_LATCH.countDown();
+        }
+
+    }
+
+    static void addApplicationProperties(JavaArchive archive, boolean server) {
+        archive.addAsResource(new StringAsset(
+                "quarkus.websockets-next." + (server ? "server" : "client") + ".traffic-logging.enabled=true\n"
+                        + "quarkus.log.category.\"io.quarkus.websockets.next.traffic\".level=DEBUG"),
+                "application.properties");
+    }
+
+    static Consumer<List<LogRecord>> logRecordsConsumer(boolean received) {
+        return recs -> {
+            assertTrue(recs.stream()
+                    .anyMatch(r -> r.getMessage().contains("%s connection opened:")));
+            assertTrue(recs.stream()
+                    .anyMatch(r -> r.getMessage()
+                            .contains("%s " + (received ? "received" : "sent") + " text message, Connection[%s]")));
+            assertTrue(recs.stream()
+                    .anyMatch(r -> r.getMessage().contains("%s connection closed,")));
+        };
+    }
+
+    static boolean isTrafficLogRecord(LogRecord r) {
+        return r.getLevel().equals(Level.FINE)
+                && r.getLoggerName().equals("io.quarkus.websockets.next.traffic");
+    }
+
+}

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

@@ -0,0 +1,20 @@
+package io.quarkus.websockets.next;
+
+import io.smallrye.config.WithDefault;
+
+public interface TrafficLoggingConfig {
+
+    /**
+     * If set to true then binary/text messages received/sent are logged if the {@code DEBUG} level is enabled for the
+     * logger {@code io.quarkus.websockets.next.traffic}.
+     */
+    @WithDefault("false")
+    public boolean enabled();
+
+    /**
+     * The number of characters of a text message which will be logged if traffic logging is enabled. The payload of a
+     * binary message is never logged.
+     */
+    @WithDefault("100")
+    public int textPayloadLimit();
+}

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

@@ -58,4 +58,9 @@ public interface WebSocketsClientRuntimeConfig {
      */
     Optional<String> tlsConfigurationName();
 
+    /**
+     * Traffic logging config.
+     */
+    TrafficLoggingConfig trafficLogging();
+
 }

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

@@ -59,6 +59,11 @@ public interface WebSocketsServerRuntimeConfig {
      */
     Security security();
 
+    /**
+     * Traffic logging config.
+     */
+    TrafficLoggingConfig trafficLogging();
+
     interface Security {
 
         /**

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

@@ -147,12 +147,15 @@ public Uni<WebSocketClientConnection> connect() {
         return UniHelper.toUni(client.connect(connectOptions))
                 .map(ws -> {
                     String clientId = BasicWebSocketConnector.class.getName();
+                    TrafficLogger trafficLogger = TrafficLogger.forClient(config);
                     WebSocketClientConnectionImpl connection = new WebSocketClientConnectionImpl(clientId, ws,
                             codecs,
                             pathParams,
                             serverEndpointUri,
-                            headers);
-                    LOG.debugf("Client connection created: %s", connection);
+                            headers, trafficLogger);
+                    if (trafficLogger != null) {
+                        trafficLogger.connectionOpened(connection);
+                    }
                     connectionManager.add(BasicWebSocketConnectorImpl.class.getName(), connection);
 
                     if (openHandler != null) {
@@ -162,8 +165,11 @@ public Uni<WebSocketClientConnection> connect() {
                     if (textMessageHandler != null) {
                         ws.textMessageHandler(new Handler<String>() {
                             @Override
-                            public void handle(String event) {
-                                doExecute(connection, event, textMessageHandler);
+                            public void handle(String message) {
+                                if (trafficLogger != null) {
+                                    trafficLogger.textMessageReceived(connection, message);
+                                }
+                                doExecute(connection, message, textMessageHandler);
                             }
                         });
                     }
@@ -172,8 +178,11 @@ public void handle(String event) {
                         ws.binaryMessageHandler(new Handler<Buffer>() {
 
                             @Override
-                            public void handle(Buffer event) {
-                                doExecute(connection, event, binaryMessageHandler);
+                            public void handle(Buffer message) {
+                                if (trafficLogger != null) {
+                                    trafficLogger.binaryMessageReceived(connection, message);
+                                }
+                                doExecute(connection, message, binaryMessageHandler);
                             }
                         });
                     }
@@ -202,6 +211,9 @@ public void handle(Throwable event) {
 
                         @Override
                         public void handle(Void event) {
+                            if (trafficLogger != null) {
+                                trafficLogger.connectionClosed(connection);
+                            }
                             if (closeHandler != null) {
                                 doExecute(connection, new CloseReason(ws.closeStatusCode(), ws.closeReason()), closeHandler);
                             }