This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch backport/27762-27764-to-camel-4.22.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit 8473ee07b7a4b3dea59b1fc1df1ade907cd0736e 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]> (cherry picked from commit 1d59394c3ecd08479864aae272f6896e5ca7e7e7) Signed-off-by: Claus Ibsen <[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"); + } + }; + } +}
