davsclaus commented on code in PR #27830:
URL: https://github.com/apache/camel/pull/27830#discussion_r4275136260
##########
components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketEndpoint.java:
##########
@@ -197,7 +251,19 @@ protected WebSocket
getWebSocket(java.util.function.Consumer<WebSocket> setupCal
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
Review Comment:
On this branch this comment is wrong, not only imprecise (see @gnodet-bot's
thread). The `onComplete` callback above (lines 236-252) does not have the
guards that #27762 added on `main`: `if (future.isDone()) { ws.close(); return;
}`, `if (!future.complete(ws)) { ws.close(); }`, and `ws.close()` when the
setup callback throws. So a connect that completes after the timeout leaves its
WebSocket open, and the consumer's handlers are already on it (`setupCallback`
runs before `future.complete`). That leak was already there before this PR.
Either port those guards with the rest of #27762, or describe here what the
code does on this branch.
##########
components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumer.java:
##########
@@ -43,22 +49,74 @@ public VertxWebsocketEndpoint getEndpoint() {
@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.
Review Comment:
On `main`, this `doStop` begins with `cancelReconnect()` (from CAMEL-25582,
#27762), which this branch does not have. Without it, a periodic reconnect
timer started by this consumer keeps running after the route is stopped
(forever with the default `maxReconnectAttempts=0`). Now that #27764 runs the
setup callback on an already-open socket, that timer can register this stopped
consumer's handlers on the WebSocket of the restarted route's new consumer (see
the next comment). Backporting #27762 first would bring in `cancelReconnect()`
as it was reviewed.
##########
components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumer.java:
##########
@@ -43,22 +49,74 @@ public VertxWebsocketEndpoint getEndpoint() {
@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();
Vertx vertx = getEndpoint().getVertx();
vertx.setPeriodic(configuration.getReconnectInitialDelay(),
configuration.getReconnectInterval(), timerId -> {
vertx.executeBlocking(() -> {
-
getEndpoint().getWebSocket(this::configureWebSocketHandlers);
+ connect(true);
Review Comment:
This tick has no `isStarted()` check, unlike `reconnect()` on `main`. If the
route was stopped and started again in the meantime, and the new consumer
already opened the WebSocket, `connect(true)` takes the `wasAlreadyOpen=true`
branch. It runs `configureWebSocketHandlers(ws)` for this stopped consumer,
which replaces the new consumer's handlers and sets `handlerOwner` to this
consumer. `mine` is null, so nothing is discarded. After that, the next
disconnect is ignored (`isStarted()` is false in this consumer's close
handler), so the route never reconnects. Stopping the route later also leaves
the handlers in place, because `clearWebSocketHandlers` is called with the new
consumer, which is no longer the owner. Before this PR,
`getWebSocket(callback)` did not run the callback on an open socket, so this
was harmless. A minimal guard is to return `null` from `connect(true)` when
`!isStarted()` after taking the lock, before `getWebSocket`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
