This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 1d59394c3ecd CAMEL-25601: camel-vertx-websocket - register client
consumer handlers on an already-open WebSocket (#27764)
1d59394c3ecd is described below
commit 1d59394c3ecd08479864aae272f6896e5ca7e7e7
Author: Guillaume Nodet <[email protected]>
AuthorDate: Wed Oct 14 13:47:24 2026 +0200
CAMEL-25601: camel-vertx-websocket - register client consumer handlers on
an already-open WebSocket (#27764)
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
---
.../websocket/VertxWebsocketClientConsumer.java | 19 +++-
.../vertx/websocket/VertxWebsocketEndpoint.java | 49 ++++++++-
.../VertxWebsocketClientConsumerHandlersTest.java | 119 +++++++++++++++++++++
3 files changed, 184 insertions(+), 3 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 d0e4b27a5d92..49d1ea8470ba 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
@@ -73,10 +73,16 @@ public class VertxWebsocketClientConsumer extends
DefaultConsumer {
Lock connectLock = getEndpoint().getClientConsumerConnectLock();
connectLock.lock();
try {
+ // Use the BiConsumer overload: getWebSocket tells the callback
whether the socket was already open.
+ // This avoids a racy external isWebSocketOpen() check: a producer
(which does not hold connectLock)
+ // could open and cache a WebSocket between that check and
getWebSocket(), causing the consumer to
+ // treat a pre-existing producer socket as its own and close it on
stop.
AtomicReference<WebSocket> opened = new AtomicReference<>();
- WebSocket webSocket = getEndpoint().getWebSocket(ws -> {
+ WebSocket webSocket = getEndpoint().getWebSocket((ws,
wasAlreadyOpen) -> {
configureWebSocketHandlers(ws);
- opened.set(ws);
+ if (!wasAlreadyOpen) {
+ opened.set(ws);
+ }
});
WebSocket mine = opened.get();
if (reconnect && mine != null && !isStarted()) {
@@ -95,10 +101,19 @@ public class VertxWebsocketClientConsumer extends
DefaultConsumer {
protected void doStop() throws Exception {
// a stopped consumer does not reconnect
cancelReconnect();
+ // Detach this consumer's handlers so a stopped consumer no longer
processes incoming messages.
+ Lock connectLock = getEndpoint().getClientConsumerConnectLock();
+ connectLock.lock();
+ try {
+ getEndpoint().clearWebSocketHandlers(this);
+ } finally {
+ connectLock.unlock();
+ }
super.doStop();
}
protected void configureWebSocketHandlers(WebSocket webSocket) {
+ getEndpoint().setHandlerOwner(this);
webSocket.binaryMessageHandler(buffer ->
this.handleResult(buffer.getBytes()));
webSocket.textMessageHandler(this::handleResult);
webSocket.closeHandler(event -> {
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 7fd77978440f..4a782648de1c 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
@@ -23,6 +23,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.BiConsumer;
import java.util.stream.Collectors;
import io.vertx.core.Vertx;
@@ -70,6 +71,8 @@ public class VertxWebsocketEndpoint extends DefaultEndpoint
implements EndpointS
private WebSocketClient client;
private WebSocketClientOptions resolvedClientOptions;
private volatile WebSocket webSocket;
+ // the consumer that last registered handlers on the endpoint's WebSocket,
guarded by clientConsumerConnectLock
+ private VertxWebsocketClientConsumer handlerOwner;
private final Lock clientConsumerConnectLock = new ReentrantLock();
public VertxWebsocketEndpoint(String uri, VertxWebsocketComponent
component, VertxWebsocketConfiguration configuration) {
@@ -163,6 +166,31 @@ public class VertxWebsocketEndpoint extends
DefaultEndpoint implements EndpointS
return clientConsumerConnectLock;
}
+ /**
+ * Clears the message, close, and exception handlers on the endpoint's
WebSocket, but only if it is still open and
+ * {@code consumer} is the current handler owner. The caller must hold
{@link #clientConsumerConnectLock}.
+ * <p>
+ * Checking the consumer rather than the socket prevents one consumer from
detaching another's handlers when two
+ * routes consume the same {@code consumeAsClient} URI and therefore share
this endpoint's WebSocket.
+ */
+ void clearWebSocketHandlers(VertxWebsocketClientConsumer consumer) {
+ if (consumer != null && handlerOwner == consumer && webSocket != null
&& !webSocket.isClosed()) {
+ webSocket.textMessageHandler(null);
+ webSocket.binaryMessageHandler(null);
+ webSocket.closeHandler(null);
+ webSocket.exceptionHandler(null);
+ handlerOwner = null;
+ }
+ }
+
+ /**
+ * Records {@code consumer} as the current handler owner on the endpoint's
WebSocket. The caller must hold
+ * {@link #clientConsumerConnectLock}.
+ */
+ void setHandlerOwner(VertxWebsocketClientConsumer consumer) {
+ this.handlerOwner = consumer;
+ }
+
protected WebSocket getWebSocket() throws Exception {
return getWebSocket((java.util.function.Consumer<WebSocket>) null);
}
@@ -172,8 +200,22 @@ public class VertxWebsocketEndpoint extends
DefaultEndpoint implements EndpointS
* {@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.
+ * <p>
+ * If the WebSocket is already open (e.g. a producer connected first, or
the consumer route was stopped and
+ * restarted while the connection was healthy), the setup callback is
still invoked on the existing socket so that
+ * the caller's handlers are registered.
*/
protected WebSocket getWebSocket(java.util.function.Consumer<WebSocket>
setupCallback) throws Exception {
+ return getWebSocket(setupCallback != null ? (ws, alreadyOpen) ->
setupCallback.accept(ws) : null);
+ }
+
+ /**
+ * Like {@link #getWebSocket(java.util.function.Consumer)}, but the setup
callback also receives a flag indicating
+ * whether the WebSocket was already open before this call ({@code true})
or was newly connected ({@code false}).
+ * This lets the caller distinguish the two cases without a racy external
check, since {@code getWebSocket} itself
+ * knows which branch it took.
+ */
+ WebSocket getWebSocket(BiConsumer<WebSocket, Boolean> setupCallback)
throws Exception {
if (client == null) {
resolvedClientOptions = configuration.getClientOptions();
if (resolvedClientOptions == null) {
@@ -202,7 +244,7 @@ public class VertxWebsocketEndpoint extends DefaultEndpoint
implements EndpointS
LOG.info("Connected to WebSocket on {}",
ws.remoteAddress());
if (setupCallback != null) {
try {
- setupCallback.accept(ws);
+ setupCallback.accept(ws, false);
} catch (Exception e) {
ws.close();
future.completeExceptionally(e);
@@ -224,6 +266,11 @@ public class VertxWebsocketEndpoint extends
DefaultEndpoint implements EndpointS
future.cancel(false);
throw e;
}
+ } else {
+ if (setupCallback != null) {
+ // WebSocket already open — register the caller's handlers on
the existing socket (CAMEL-25601).
+ setupCallback.accept(webSocket, true);
+ }
}
return webSocket;
}
diff --git
a/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumerHandlersTest.java
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumerHandlersTest.java
new file mode 100644
index 000000000000..7e6f2208e51d
--- /dev/null
+++
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumerHandlersTest.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.vertx.websocket;
+
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies that a client consumer started against an already-open WebSocket
(CAMEL-25601) correctly registers its
+ * message, close, and exception handlers, and that stopping the consumer
detaches them so no messages leak through.
+ * <p>
+ * Two scenarios:
+ * <ol>
+ * <li>The consumer route is stopped and restarted while the server is still
up and the connection is healthy; the
+ * restarted consumer must receive messages (previously it registered no
handlers).</li>
+ * <li>A producer on the same {@code consumeAsClient=true} endpoint opens the
WebSocket first; a consumer started
+ * afterward must register its handlers on that already-open socket.</li>
+ * </ol>
+ */
+public class VertxWebsocketClientConsumerHandlersTest extends
VertxWebSocketTestSupport {
+
+ /**
+ * Scenario 1: client consumer route stopped then restarted while the
WebSocket stays open.
+ * <p>
+ * When a route is stopped and restarted (while the server is healthy),
the endpoint is not stopped and the
+ * WebSocket stays open. The restarted consumer must register its handlers
on the existing socket and continue to
+ * receive messages.
+ */
+ @Test
+ void clientConsumerReceivesMessagesAfterRouteStopAndRestart() throws
Exception {
+ String serverUri = String.format("vertx-websocket:localhost:%d/echo",
port.getPort());
+
+ // Start the client-consumer route first (it uses noAutoStartup).
+ context.getRouteController().startRoute("client-consumer");
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("before-restart");
+ template.sendBody(serverUri, "before-restart");
+ mock.assertIsSatisfied();
+
+ // Stop the consumer route. The WebSocket stays open (the endpoint is
not stopped).
+ context.getRouteController().stopRoute("client-consumer");
+
+ // While the route is stopped, messages must NOT be delivered to the
mock (stop half of CAMEL-25601).
+ mock.reset();
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(500);
+ template.sendBody(serverUri, "while-stopped");
+ mock.assertIsSatisfied();
+
+ // Restart the consumer route. The WebSocket is already open — the fix
ensures handlers are re-registered.
+ context.getRouteController().startRoute("client-consumer");
+
+ mock.reset();
+ mock.expectedBodiesReceived("after-restart");
+ template.sendBody(serverUri, "after-restart");
+ mock.assertIsSatisfied();
+ }
+
+ /**
+ * Scenario 2: consumer started after a producer on the same endpoint has
already opened the WebSocket.
+ * <p>
+ * A producer sending via the {@code consumeAsClient=true} endpoint opens
(and caches) the client WebSocket on that
+ * endpoint. A consumer route started afterward must register its handlers
on the same already-open socket.
+ */
+ @Test
+ void clientConsumerReceivesMessagesWhenStartedAfterProducerOpenedSocket()
throws Exception {
+ String serverUri = String.format("vertx-websocket:localhost:%d/echo",
port.getPort());
+ String clientUri =
String.format("vertx-websocket:localhost:%d/echo?consumeAsClient=true",
port.getPort());
+
+ // Send via the consumeAsClient endpoint to open its WebSocket before
the consumer route starts.
+ // The server echoes the message; this reply arrives on the socket we
just opened.
+ template.sendBody(clientUri, "warm-up");
+
+ // Now start the client-consumer route: the WebSocket on the
consumeAsClient endpoint is already open.
+ context.getRouteController().startRoute("client-consumer");
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMinimumMessageCount(1);
+ // Send another message so the server echoes it to the consumer.
+ template.sendBody(serverUri, "after-consumer-start");
+ mock.assertIsSatisfied();
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // Echo server: reflects every message back to all connected
clients.
+ fromF("vertx-websocket:localhost:%d/echo",
port.getPort()).routeId("server")
+
.toF("vertx-websocket:localhost:%d/echo?sendToAll=true", port.getPort());
+
+ // Client consumer: connects as a WebSocket client and
forwards received messages to the mock.
+ // noAutoStartup=true so the tests can control when it
connects.
+
fromF("vertx-websocket:localhost:%d/echo?consumeAsClient=true", port.getPort())
+ .routeId("client-consumer")
+ .noAutoStartup()
+ .to("mock:result");
+ }
+ };
+ }
+}
