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 466846cbf2b46f79217749acf7bb8db25fe60bf2 Author: allthingssecurity <[email protected]> AuthorDate: Tue Oct 13 12:26:23 2026 +0530 CAMEL-25582: camel-vertx-websocket - Reconnect the client consumer with one attempt at a time (#27762) Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Co-Authored-By: Claude Opus 5.5 <[email protected]> (cherry picked from commit 05b3e67bb9f28441dd25c24ec61c4d508ebd4ff2) Signed-off-by: Claus Ibsen <[email protected]> --- .../websocket/VertxWebsocketClientConsumer.java | 172 +++++++++++--- .../vertx/websocket/VertxWebsocketEndpoint.java | 34 ++- ...WebsocketConsumerAsClientSlowReconnectTest.java | 250 +++++++++++++++++++++ 3 files changed, 421 insertions(+), 35 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..d0e4b27a5d92 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 @@ -16,7 +16,11 @@ */ package org.apache.camel.component.vertx.websocket; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import io.vertx.core.Vertx; import io.vertx.core.http.WebSocket; @@ -31,6 +35,15 @@ import org.slf4j.LoggerFactory; public class VertxWebsocketClientConsumer extends DefaultConsumer { private static final Logger LOG = LoggerFactory.getLogger(VertxWebsocketClientConsumer.class); + private static final long NO_TIMER = -1; + + private final Lock reconnectLock = new ReentrantLock(); + private final AtomicBoolean reconnecting = new AtomicBoolean(); + // id of the periodic reconnect timer, guarded by reconnectLock + private long reconnectTimerId = NO_TIMER; + // 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,44 +56,55 @@ 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 { + AtomicReference<WebSocket> opened = new AtomicReference<>(); + WebSocket webSocket = getEndpoint().getWebSocket(ws -> { + configureWebSocketHandlers(ws); + 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 { + // a stopped consumer does not reconnect + cancelReconnect(); + super.doStop(); } protected void configureWebSocketHandlers(WebSocket webSocket) { 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(); - - Vertx vertx = getEndpoint().getVertx(); - vertx.setPeriodic(configuration.getReconnectInitialDelay(), configuration.getReconnectInterval(), timerId -> { - vertx.executeBlocking(() -> { - getEndpoint().getWebSocket(this::configureWebSocketHandlers); - vertx.cancelTimer(timerId); - return null; - }, false) - .onComplete(result -> { - if (result.failed()) { - Throwable cause = result.cause(); - if (cause != null) { - LOG.debug("WebSocket reconnect to {} failed due to {}", webSocket.remoteAddress(), - cause); - } - - if (configuration.getMaxReconnectAttempts() > 0) { - if (reconnectAttempts.incrementAndGet() == configuration.getMaxReconnectAttempts()) { - LOG.warn( - "Reconnect max attempts ({}) exhausted. Giving up trying to reconnect to {}", - configuration.getMaxReconnectAttempts(), webSocket.remoteAddress()); - vertx.cancelTimer(timerId); - } - } - } - }); - }); + scheduleReconnect(webSocket); } }); webSocket.exceptionHandler(exception -> { @@ -95,6 +119,92 @@ public class VertxWebsocketClientConsumer extends DefaultConsumer { }); } + /** + * Starts the periodic reconnect, unless it is already running: a consumer has one reconnect timer, whatever the + * number of close events it gets. + */ + private void scheduleReconnect(WebSocket closedWebSocket) { + reconnectLock.lock(); + try { + if (reconnectTimerId != NO_TIMER || !isStarted()) { + return; + } + VertxWebsocketConfiguration configuration = getEndpoint().getConfiguration(); + AtomicInteger reconnectAttempts = new AtomicInteger(); + reconnectTimerId = getEndpoint().getVertx().setPeriodic(configuration.getReconnectInitialDelay(), + configuration.getReconnectInterval(), + timerId -> reconnect(timerId, closedWebSocket, configuration, reconnectAttempts)); + } finally { + reconnectLock.unlock(); + } + } + + private void reconnect( + long timerId, WebSocket closedWebSocket, VertxWebsocketConfiguration configuration, + AtomicInteger reconnectAttempts) { + if (!isStarted()) { + cancelReconnect(timerId); + return; + } + // An attempt can take longer than the reconnect interval (up to the connect timeout). Attempts that + // overlap would each open a WebSocket with the handlers of this consumer. + if (!reconnecting.compareAndSet(false, true)) { + return; + } + Vertx vertx = getEndpoint().getVertx(); + vertx.executeBlocking(() -> connect(true), false) + .onComplete(result -> { + reconnecting.set(false); + if (result.succeeded()) { + cancelReconnect(timerId); + // null: the attempt closed its WebSocket because the consumer was stopped + if (result.result() != null && result.result().isClosed()) { + // closed again while its close event was ignored because this timer was still running + scheduleReconnect(result.result()); + } + } else { + Throwable cause = result.cause(); + if (cause != null) { + LOG.debug("WebSocket reconnect to {} failed due to {}", closedWebSocket.remoteAddress(), + cause); + } + + if (configuration.getMaxReconnectAttempts() > 0) { + if (reconnectAttempts.incrementAndGet() == configuration.getMaxReconnectAttempts()) { + LOG.warn( + "Reconnect max attempts ({}) exhausted. Giving up trying to reconnect to {}", + configuration.getMaxReconnectAttempts(), closedWebSocket.remoteAddress()); + cancelReconnect(timerId); + } + } + } + }); + } + + private void cancelReconnect() { + reconnectLock.lock(); + try { + if (reconnectTimerId != NO_TIMER) { + getEndpoint().getVertx().cancelTimer(reconnectTimerId); + reconnectTimerId = NO_TIMER; + } + } finally { + reconnectLock.unlock(); + } + } + + private void cancelReconnect(long timerId) { + reconnectLock.lock(); + try { + getEndpoint().getVertx().cancelTimer(timerId); + if (reconnectTimerId == timerId) { + reconnectTimerId = NO_TIMER; + } + } finally { + reconnectLock.unlock(); + } + } + protected void handleResult(Object result) { Exchange exchange = createExchange(false); Message message = exchange.getMessage(); 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..7fd77978440f 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,8 @@ 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.stream.Collectors; import io.vertx.core.Vertx; @@ -67,7 +69,8 @@ public class VertxWebsocketEndpoint extends DefaultEndpoint implements EndpointS private WebSocketClient client; private WebSocketClientOptions resolvedClientOptions; - private WebSocket webSocket; + private volatile WebSocket webSocket; + private final Lock clientConsumerConnectLock = new ReentrantLock(); public VertxWebsocketEndpoint(String uri, VertxWebsocketComponent component, VertxWebsocketConfiguration configuration) { super(uri, component); @@ -151,6 +154,15 @@ 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; + } + protected WebSocket getWebSocket() throws Exception { return getWebSocket((java.util.function.Consumer<WebSocket>) null); } @@ -182,22 +194,36 @@ public class VertxWebsocketEndpoint extends DefaultEndpoint implements EndpointS client.connect(connectOptions).onComplete(result -> { if (result.succeeded()) { WebSocket ws = result.result(); + if (future.isDone()) { + // the caller gave up waiting (see below): nobody would use or close this WebSocket + ws.close(); + return; + } LOG.info("Connected to WebSocket on {}", ws.remoteAddress()); if (setupCallback != null) { try { setupCallback.accept(ws); } catch (Exception e) { + ws.close(); future.completeExceptionally(e); return; } } - future.complete(ws); + if (!future.complete(ws)) { + ws.close(); + } } else { - webSocket = null; 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; + } } return webSocket; } diff --git a/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketConsumerAsClientSlowReconnectTest.java b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketConsumerAsClientSlowReconnectTest.java new file mode 100644 index 000000000000..834515814a48 --- /dev/null +++ b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketConsumerAsClientSlowReconnectTest.java @@ -0,0 +1,250 @@ +/* + * 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 java.time.Duration; +import java.util.List; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import io.vertx.core.Vertx; +import io.vertx.core.http.HttpServer; +import io.vertx.core.http.ServerWebSocket; +import io.vertx.core.http.ServerWebSocketHandshake; +import org.apache.camel.Consumer; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.StatefulService; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertAll; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Reconnects of the client consumer against a server whose answer takes longer than the reconnect interval, or that + * rejects the handshake. The server is a plain Vert.x HTTP server that counts the handshakes and the open WebSockets. + */ +@Timeout(90) +class VertxWebsocketConsumerAsClientSlowReconnectTest extends VertxWebSocketTestSupport { + + private static final int RECONNECT_INTERVAL = 50; + // several reconnect intervals + private static final int HANDSHAKE_DELAY = 1000; + + private enum Mode { + ACCEPT, + ACCEPT_LATE, + HOLD, + REJECT + } + + private final Set<ServerWebSocket> open = ConcurrentHashMap.newKeySet(); + private final AtomicInteger handshakes = new AtomicInteger(); + private final AtomicInteger pending = new AtomicInteger(); + // WebSockets the server has opened, including the ones that are closed again + private final AtomicInteger opened = new AtomicInteger(); + // handshakes of the HOLD mode, answered when the test says so + private final Queue<ServerWebSocketHandshake> held = new ConcurrentLinkedQueue<>(); + private volatile Mode mode = Mode.ACCEPT; + + private Vertx serverVertx; + + @Override + protected void setupResources() throws Exception { + serverVertx = Vertx.vertx(); + HttpServer server = serverVertx.createHttpServer(); + server.webSocketHandshakeHandler(handshake -> { + handshakes.incrementAndGet(); + switch (mode) { + case ACCEPT -> handshake.accept(); + case REJECT -> handshake.reject(); + case HOLD -> held.add(handshake); + case ACCEPT_LATE -> { + pending.incrementAndGet(); + serverVertx.setTimer(HANDSHAKE_DELAY, + timerId -> handshake.accept().onComplete(done -> pending.decrementAndGet())); + } + } + }); + server.webSocketHandler(webSocket -> { + open.add(webSocket); + opened.incrementAndGet(); + webSocket.closeHandler(closed -> open.remove(webSocket)); + }); + server.listen(port.getPort(), "localhost").toCompletionStage().toCompletableFuture().get(10, TimeUnit.SECONDS); + } + + @Override + protected void cleanupResources() throws Exception { + if (serverVertx != null) { + serverVertx.close().toCompletionStage().toCompletableFuture().get(10, TimeUnit.SECONDS); + } + } + + @Test + void reconnectSlowerThanTheIntervalOpensOneWebSocket() throws Exception { + awaitOpenWebSockets(1); + + // the connection is lost, and the server answers a handshake only after several reconnect intervals + mode = Mode.ACCEPT_LATE; + int before = handshakes.get(); + closeOpenWebSockets(); + + // all the handshakes the consumer sent are answered + await().atMost(30, TimeUnit.SECONDS) + .until(() -> handshakes.get() > before && pending.get() == 0 && !open.isEmpty()); + + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("Hello"); + mock.setAssertPeriod(1000); + List.copyOf(open).forEach(webSocket -> webSocket.writeTextMessage("Hello")); + + assertAll( + () -> assertEquals(1, handshakes.get() - before, "handshakes sent for one reconnect"), + () -> assertEquals(1, open.size(), "WebSockets the consumer has open after one reconnect"), + () -> mock.assertIsSatisfied()); + } + + @Test + void reconnectThatCompletesAfterStopIsClosed() throws Exception { + awaitOpenWebSockets(1); + + mode = Mode.ACCEPT_LATE; + closeOpenWebSockets(); + // a reconnect is in flight + await().atMost(30, TimeUnit.SECONDS).until(() -> pending.get() > 0); + + context.getRouteController().stopRoute("client"); + + await().atMost(30, TimeUnit.SECONDS).untilAsserted( + () -> assertEquals(0, pending.get() + open.size(), + "WebSockets pending or open after the consumer was stopped")); + } + + @Test + void stoppedConsumerDoesNotReconnect() throws Exception { + awaitOpenWebSockets(1); + + // the consumer retries every 50 ms while the server rejects the handshake + mode = Mode.REJECT; + int before = handshakes.get(); + closeOpenWebSockets(); + await().atMost(30, TimeUnit.SECONDS).until(() -> handshakes.get() >= before + 3); + + context.getRouteController().stopRoute("client"); + + // no handshake for 20 reconnect intervals + AtomicInteger seen = new AtomicInteger(-1); + await("no reconnect attempt after the consumer was stopped") + .during(Duration.ofSeconds(1)).atMost(Duration.ofSeconds(10)) + .until(() -> seen.getAndSet(handshakes.get()) == handshakes.get()); + } + + @Test + void restartWhileReconnectIsPendingOpensOneWebSocket() throws Exception { + awaitOpenWebSockets(1); + + // the connection is lost, and the server does not answer the handshake of the reconnect yet + mode = Mode.HOLD; + int handshakesBefore = handshakes.get(); + int openedBefore = opened.get(); + closeOpenWebSockets(); + await().atMost(30, TimeUnit.SECONDS).until(() -> !held.isEmpty()); + + // the route is stopped and started again (with a new consumer) while that attempt waits for the answer + Consumer stopped = context.getRoute("client").getConsumer(); + context.getRouteController().stopRoute("client"); + CompletableFuture<Void> restart = CompletableFuture.runAsync(() -> { + try { + context.getRouteController().startRoute("client"); + } catch (Exception e) { + throw new CompletionException(e); + } + }); + try { + await().atMost(30, TimeUnit.SECONDS) + .until(() -> isStarting(context.getRoute("client").getConsumer(), stopped) || restart.isDone()); + + // now the server answers: the pending handshake(s), and at once from here on + mode = Mode.ACCEPT; + for (ServerWebSocketHandshake handshake = held.poll(); handshake != null; handshake = held.poll()) { + handshake.accept(); + } + restart.get(30, TimeUnit.SECONDS); + } finally { + // never leave a start waiting for a handshake that is not answered + mode = Mode.ACCEPT; + for (ServerWebSocketHandshake handshake = held.poll(); handshake != null; handshake = held.poll()) { + handshake.accept(); + } + } + + // The pending attempt belongs to the stopped consumer: it closes its WebSocket, and the start of the new + // consumer, which waited for that attempt, connects. Two handshakes, one WebSocket left. + await().atMost(30, TimeUnit.SECONDS).untilAsserted( + () -> assertAll( + () -> assertEquals(2, handshakes.get() - handshakesBefore, + "handshakes sent by the reconnect and the restart"), + () -> assertEquals(2, opened.get() - openedBefore, + "WebSockets opened by the reconnect and the restart"), + () -> assertEquals(1, open.size(), "WebSockets the consumer has open after the restart"))); + + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("Hello"); + mock.setAssertPeriod(1000); + List.copyOf(open).forEach(webSocket -> webSocket.writeTextMessage("Hello")); + mock.assertIsSatisfied(); + + assertAll( + () -> assertEquals(2, handshakes.get() - handshakesBefore, "no further handshake after the restart"), + () -> assertEquals(1, open.size(), "WebSockets the consumer has open after the message")); + } + + private static boolean isStarting(Consumer consumer, Consumer stopped) { + return consumer != stopped && consumer instanceof StatefulService service && service.isStarting(); + } + + private void awaitOpenWebSockets(int expected) { + await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(expected, open.size())); + } + + private void closeOpenWebSockets() { + List.copyOf(open).forEach(ServerWebSocket::close); + } + + @Override + protected RoutesBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + fromF("vertx-websocket:localhost:%d/feed?consumeAsClient=true&reconnectInterval=%d", port.getPort(), + RECONNECT_INTERVAL) + .routeId("client") + .to("mock:result"); + } + }; + } +}
