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 05b3e67bb9f2 CAMEL-25582: camel-vertx-websocket - Reconnect the client
consumer with one attempt at a time (#27762)
05b3e67bb9f2 is described below
commit 05b3e67bb9f28441dd25c24ec61c4d508ebd4ff2
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]>
---
.../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");
+ }
+ };
+ }
+}
