From ce035483e1961314afc8f9ff046a9736f244d3c4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aur=C3=A9lien=20Pupier?= <apupier@ibm.com>
Date: Wed, 15 Jul 2026 17:26:29 +0200
Subject: [PATCH] CAMEL-24126 - Ensure no message sent before the
 VertxWebsocket is fully configured
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

it was visible with flaky test
VertxWebsocketHandshakeHeadersTest.testHandshakeHeadersAsConsumer
see rtx-websocket:surefire:test@default-test/details/org.apache.camel.component.vertx.websocket.VertxWebsocketHandshakeHeadersTest/testHandshakeHeadersAsConsumer?top-execution=1

Co-authored-by: IBM Bob IDE 2.0.1
Signed-off-by: Aurélien Pupier <apupier@ibm.com>
---
 .../VertxWebsocketClientConsumer.java         |  4 ++--
 .../websocket/VertxWebsocketEndpoint.java     | 23 +++++++++++++++++--
 2 files changed, 23 insertions(+), 4 deletions(-)

diff --git a/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumer.java b/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumer.java
index 0cb287314e8f1..6d168037f02fa 100644
--- a/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumer.java
+++ b/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumer.java
@@ -43,7 +43,7 @@ public VertxWebsocketEndpoint getEndpoint() {
 
     @Override
     protected void doStart() throws Exception {
-        configureWebSocketHandlers(getEndpoint().getWebSocket());
+        getEndpoint().getWebSocket(this::configureWebSocketHandlers);
     }
 
     protected void configureWebSocketHandlers(WebSocket webSocket) {
@@ -58,7 +58,7 @@ protected void configureWebSocketHandlers(WebSocket webSocket) {
                 Vertx vertx = getEndpoint().getVertx();
                 vertx.setPeriodic(configuration.getReconnectInitialDelay(), configuration.getReconnectInterval(), timerId -> {
                     vertx.executeBlocking(() -> {
-                        configureWebSocketHandlers(getEndpoint().getWebSocket());
+                        getEndpoint().getWebSocket(this::configureWebSocketHandlers);
                         vertx.cancelTimer(timerId);
                         return null;
                     }, false)
diff --git a/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketEndpoint.java b/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketEndpoint.java
index 7342c778f699b..3467b94b5a5fe 100644
--- a/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketEndpoint.java
+++ b/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketEndpoint.java
@@ -152,6 +152,16 @@ protected Vertx getVertx() {
     }
 
     protected WebSocket getWebSocket() throws Exception {
+        return getWebSocket((java.util.function.Consumer<WebSocket>) null);
+    }
+
+    /**
+     * Connects to the WebSocket server, optionally invoking a setup callback on the socket <em>before</em> the blocking
+     * {@code future.get()} returns. This guarantees that any message/close/exception handlers supplied by the callback
+     * are registered atomically with the connection, so no messages sent by the server immediately after the handshake
+     * can be missed.
+     */
+    protected WebSocket getWebSocket(java.util.function.Consumer<WebSocket> setupCallback) throws Exception {
         if (client == null) {
             resolvedClientOptions = configuration.getClientOptions();
             if (resolvedClientOptions == null) {
@@ -171,8 +181,17 @@ protected WebSocket getWebSocket() throws Exception {
             CompletableFuture<WebSocket> future = new CompletableFuture<>();
             client.connect(connectOptions).onComplete(result -> {
                 if (result.succeeded()) {
-                    LOG.info("Connected to WebSocket on {}", result.result().remoteAddress());
-                    future.complete(result.result());
+                    WebSocket ws = result.result();
+                    LOG.info("Connected to WebSocket on {}", ws.remoteAddress());
+                    if (setupCallback != null) {
+                        try {
+                            setupCallback.accept(ws);
+                        } catch (Exception e) {
+                            future.completeExceptionally(e);
+                            return;
+                        }
+                    }
+                    future.complete(ws);
                 } else {
                     webSocket = null;
                     future.completeExceptionally(result.cause());