Propagate correctly the HTTP connection metric to the WebSocket connection metric during the server upgrade.

Motivation:

During a server WebSocket upgrade, the HTTP connection metric is not correctly set to the WebSocket connection metric.

This leads to calling the metric implementation with a null instance of the metric instead of the the instance it was provided.

Changes:

During the server WebSocket upgrade set the HTTP connection metric onto the WebSocket connection metric.

Author: vietj

vertx-core/src/main/java/io/vertx/core/http/impl/ServerWebSocketHandshaker.java

@@ -177,7 +177,7 @@ private ServerWebSocket acceptHandshake() {
       long closingTimeoutMS = options.getWebSocketClosingTimeout() >= 0 ? options.getWebSocketClosingTimeout() * 1000L : 0L;
       WebSocketConnectionImpl webSocketConn = new WebSocketConnectionImpl(request.context, ctx, true, closingTimeoutMS,httpConn.metrics);
       ServerWebSocketImpl webSocket = new ServerWebSocketImpl(
-        (ContextInternal) request.context(),
+        request.context(),
         webSocketConn,
         handshaker.version() != WebSocketVersion.V00,
         request,
@@ -187,7 +187,7 @@ private ServerWebSocket acceptHandshake() {
       String subprotocol = handshaker.selectedSubprotocol();
       webSocket.subProtocol(subprotocol);
       webSocketConn.webSocket(webSocket);
-      webSocketConn.metric(webSocketConn.metric());
+      webSocketConn.metric(httpConn.metric());
       return webSocketConn;
     });
     CompletableFuture<Void> latch = new CompletableFuture<>();

vertx-core/src/test/java/io/vertx/tests/metrics/MetricsTest.java

@@ -44,6 +44,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
@@ -564,11 +565,17 @@ public void testServerWebSocket() throws InterruptedException {
     server.webSocketHandler(ws -> {
       wsRef.set(ws);
       FakeHttpServerMetrics metrics = FakeMetricsBase.getMetrics(server);
-      WebSocketMetric metric = metrics.getWebSocketMetric(ws);
-      assertNotNull(metric);
+      WebSocketMetric webSocketMetric = metrics.getWebSocketMetric(ws);
+      SocketMetric socketMetric = metrics.firstMetric(ws.remoteAddress());
+      long bytesWritten = socketMetric.bytesRead.get();
+      long bytesRead = socketMetric.bytesRead.get();
+      assertNotNull(webSocketMetric);
       ws.handler(ws::write);
       ws.closeHandler(closed -> {
-        latch.countDown();
+        vertx.runOnContext(v -> {
+          assertTrue(socketMetric.bytesRead.get() > bytesRead || socketMetric.bytesWritten.get() > bytesWritten);
+          latch.countDown();
+        });
       });
     });
     awaitFuture(server.listen(HttpTestBase.DEFAULT_HTTP_PORT, HttpTestBase.DEFAULT_HTTP_HOST));
@@ -589,15 +596,23 @@ public void testServerWebSocketUpgrade() throws InterruptedException {
     server.requestHandler(req -> {
       FakeHttpServerMetrics metrics = FakeMetricsBase.getMetrics(server);
       assertNotNull(metrics.getRequestMetric(req));
-      req.toWebSocket().onComplete(onSuccess(ws -> {
-        assertNull(metrics.getRequestMetric(req));
-        WebSocketMetric metric = metrics.getWebSocketMetric(ws);
-        assertNotNull(metric);
-        ws.handler(ws::write);
-        ws.closeHandler(closed -> {
-          ref.set(ws);
-        });
-      }));
+      req
+        .toWebSocket()
+        .onComplete(onSuccess(ws -> {
+          assertNull(metrics.getRequestMetric(req));
+          WebSocketMetric wsMetric = metrics.getWebSocketMetric(ws);
+          SocketMetric socketMetric = metrics.firstMetric(ws.remoteAddress());
+          long bytesWritten = socketMetric.bytesRead.get();
+          long bytesRead = socketMetric.bytesRead.get();
+          assertNotNull(wsMetric);
+          ws.handler(ws::write);
+          ws.closeHandler(closed -> {
+            vertx.runOnContext(v -> {
+              assertTrue(socketMetric.bytesRead.get() > bytesRead || socketMetric.bytesWritten.get() > bytesWritten);
+              ref.set(ws);
+            });
+          });
+        }));
     });
     awaitFuture(server.listen(HttpTestBase.DEFAULT_HTTP_PORT, HttpTestBase.DEFAULT_HTTP_HOST));
     wsClient = vertx.createWebSocketClient();