This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch backport/27764-to-camel-4.22.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit b6e6221837da0e92458e9bc07120745bc0752e44 Author: Guillaume Nodet <[email protected]> AuthorDate: Wed Oct 14 11:54:37 2026 +0000 [backport camel-4.22.x] CAMEL-25601: camel-vertx-websocket - register client consumer handlers on an already-open WebSocket --- .../websocket/VertxWebsocketClientConsumer.java | 64 ++++++++++- .../vertx/websocket/VertxWebsocketEndpoint.java | 72 ++++++++++++- .../VertxWebsocketClientConsumerHandlersTest.java | 119 +++++++++++++++++++++ 3 files changed, 249 insertions(+), 6 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 6d168037f02f..8cb7d41d5074 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 @@ -17,6 +17,8 @@ package org.apache.camel.component.vertx.websocket; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Lock; import io.vertx.core.Vertx; import io.vertx.core.http.WebSocket; @@ -32,6 +34,10 @@ import org.slf4j.LoggerFactory; public class VertxWebsocketClientConsumer extends DefaultConsumer { private static final Logger LOG = LoggerFactory.getLogger(VertxWebsocketClientConsumer.class); + // the WebSocket that a reconnect attempt closed because the consumer was stopped: its close event is not a lost + // connection + private volatile WebSocket discarded; + public VertxWebsocketClientConsumer(Endpoint endpoint, Processor processor) { super(endpoint, processor); } @@ -43,14 +49,66 @@ public class VertxWebsocketClientConsumer extends DefaultConsumer { @Override protected void doStart() throws Exception { - getEndpoint().getWebSocket(this::configureWebSocketHandlers); + connect(false); + } + + /** + * Returns the WebSocket of the endpoint, connected with the handlers of this consumer. One call at a time for all + * the client consumers of the endpoint (a route that is stopped and started again gets a new consumer): a start and + * a reconnect attempt that both find no open WebSocket would each open one. + * + * @param reconnect whether this is a reconnect attempt. A reconnect attempt that opened a WebSocket and finds the + * consumer no longer started closes that WebSocket and returns {@code null}. This is done while + * the lock is held: a start that waits for the lock then finds the WebSocket closed and connects + * again, instead of taking over a WebSocket that is about to be closed. + */ + private WebSocket connect(boolean reconnect) throws Exception { + 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, wasAlreadyOpen) -> { + configureWebSocketHandlers(ws); + if (!wasAlreadyOpen) { + opened.set(ws); + } + }); + WebSocket mine = opened.get(); + if (reconnect && mine != null && !isStarted()) { + // Stopped while connecting. Not for a WebSocket that was already open: it belongs to the endpoint. + discarded = mine; + mine.close(); + return null; + } + return webSocket; + } finally { + connectLock.unlock(); + } + } + + @Override + protected void doStop() throws Exception { + // 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 -> { - if (isStarted()) { + if (isStarted() && webSocket != discarded) { LOG.info("WebSocket disconnected from {}. Attempting to reconnect...", webSocket.remoteAddress()); VertxWebsocketConfiguration configuration = getEndpoint().getConfiguration(); AtomicInteger reconnectAttempts = new AtomicInteger(); @@ -58,7 +116,7 @@ public class VertxWebsocketClientConsumer extends DefaultConsumer { Vertx vertx = getEndpoint().getVertx(); vertx.setPeriodic(configuration.getReconnectInitialDelay(), configuration.getReconnectInterval(), timerId -> { vertx.executeBlocking(() -> { - getEndpoint().getWebSocket(this::configureWebSocketHandlers); + connect(true); 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 3467b94b5a5f..4bd5235c0f26 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 @@ -21,6 +21,9 @@ import java.util.Arrays; import java.util.Map; 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; @@ -67,7 +70,10 @@ public class VertxWebsocketEndpoint extends DefaultEndpoint implements EndpointS private WebSocketClient client; private WebSocketClientOptions resolvedClientOptions; - private WebSocket webSocket; + 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) { super(uri, component); @@ -151,6 +157,40 @@ public class VertxWebsocketEndpoint extends DefaultEndpoint implements EndpointS return getComponent().getVertx(); } + /** + * The lock under which the client consumers of this endpoint connect ({@code consumeAsClient=true}). It belongs to + * the endpoint and not to the consumer, because a route that is stopped and started again gets a new consumer while + * a reconnect attempt of the previous one can still be waiting for its handshake. + */ + Lock getClientConsumerConnectLock() { + 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); } @@ -160,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) { @@ -185,7 +239,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) { future.completeExceptionally(e); return; @@ -197,7 +251,19 @@ public class VertxWebsocketEndpoint extends DefaultEndpoint implements EndpointS future.completeExceptionally(result.cause()); } }); - webSocket = future.get(resolvedClientOptions.getConnectTimeout(), TimeUnit.MILLISECONDS); + try { + webSocket = future.get(resolvedClientOptions.getConnectTimeout(), TimeUnit.MILLISECONDS); + } catch (Exception e) { + webSocket = null; + // a connect that completes from now on closes its WebSocket + 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"); + } + }; + } +}
