davsclaus commented on code in PR #27762:
URL: https://github.com/apache/camel/pull/27762#discussion_r4255126660
##########
components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumer.java:
##########
@@ -43,44 +56,55 @@ 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 {
+ 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;
Review Comment:
Not introduced here, but worth noting for a follow-up: when `getWebSocket`
finds the endpoint's WebSocket already open, the setup callback is not called,
so this consumer never registers its handlers.
That happens on a plain `stopRoute`/`startRoute` while the connection is
healthy. The endpoint is not stopped, so the WebSocket stays open and the new
consumer gets no handlers. The stopped consumer's handlers stay on the
WebSocket and keep calling `handleResult` → `getProcessor().process(...)`, even
though that consumer is stopped. It also happens when a producer on the same
endpoint opened the WebSocket first.
A follow-up could register the handlers on an already-open WebSocket in
`doStart`, or detach them in `doStop`.
--
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]
