This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 01087d5c21ef [backport camel-4.22.x] CAMEL-25582, CAMEL-25601:
camel-vertx-websocket - reconnect the client consumer one attempt at a time,
and register its handlers on an already-open WebSocket (#27904)
01087d5c21ef is described below
commit 01087d5c21efae45811e5e4380ac9ca251ec9a5e
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Oct 15 20:39:11 2026 +0200
[backport camel-4.22.x] CAMEL-25582, CAMEL-25601: camel-vertx-websocket -
reconnect the client consumer one attempt at a time, and register its handlers
on an already-open WebSocket (#27904)
Backport of #27762 (CAMEL-25582) and #27764 (CAMEL-25601) to camel-4.22.x.
* CAMEL-25582: camel-vertx-websocket - Reconnect the client consumer with
one attempt at a time (#27762)
* CAMEL-25601: camel-vertx-websocket - register client consumer handlers on
an already-open WebSocket (#27764)
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-authored-by: Claude Sonnet 4.6 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../websocket/VertxWebsocketClientConsumer.java | 187 ++++++++++++---
.../vertx/websocket/VertxWebsocketEndpoint.java | 83 ++++++-
.../VertxWebsocketClientConsumerHandlersTest.java | 119 ++++++++++
...WebsocketConsumerAsClientSlowReconnectTest.java | 250 +++++++++++++++++++++
4 files changed, 603 insertions(+), 36 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..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
@@ -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,70 @@ 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 {
+ // 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 -> {
- 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 +134,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..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
@@ -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) {
@@ -182,22 +236,41 @@ 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);
+ setupCallback.accept(ws, false);
} 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;
+ }
+ } 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");
+ }
+ };
+ }
+}
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");
+ }
+ };
+ }
+}
