(camel) branch main updated: CAMEL-24126 - Ensure no message sent before the VertxWebsocket is fully configured

[apupier]

This is an automated email from the ASF dual-hosted git repository.

apupier pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/main by this push:
     new 4d93dcf931c5 CAMEL-24126 - Ensure no message sent before the 
VertxWebsocket is fully configured
4d93dcf931c5 is described below

commit 4d93dcf931c5b37b19f884da125faf1702686f8a
Author: Aurélien Pupier <apup...@ibm.com>
AuthorDate: Wed Jul 15 17:26:29 2026 +0200

    CAMEL-24126 - Ensure no message sent before the VertxWebsocket is fully
    configured
    
    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 <apup...@ibm.com>
---
 .../websocket/VertxWebsocketClientConsumer.java    |  4 ++--
 .../vertx/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 0cb287314e8f..6d168037f02f 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 class VertxWebsocketClientConsumer extends 
DefaultConsumer {
 
     @Override
     protected void doStart() throws Exception {
-        configureWebSocketHandlers(getEndpoint().getWebSocket());
+        getEndpoint().getWebSocket(this::configureWebSocketHandlers);
     }
 
     protected void configureWebSocketHandlers(WebSocket webSocket) {
@@ -58,7 +58,7 @@ public class VertxWebsocketClientConsumer extends 
DefaultConsumer {
                 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 7342c778f699..3467b94b5a5f 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 @@ public class VertxWebsocketEndpoint extends 
DefaultEndpoint implements EndpointS
     }
 
     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 @@ public class VertxWebsocketEndpoint extends 
DefaultEndpoint implements EndpointS
             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());