davsclaus commented on code in PR #27764:
URL: https://github.com/apache/camel/pull/27764#discussion_r4257436146
##########
components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumer.java:
##########
@@ -73,10 +73,16 @@ private WebSocket connect(boolean reconnect) throws
Exception {
Lock connectLock = getEndpoint().getClientConsumerConnectLock();
connectLock.lock();
try {
+ // Check before connecting: if the WebSocket is already open, the
setup callback will be invoked on it
+ // (see getWebSocket javadoc), but "opened" must only track
newly-opened sockets so that the reconnect
+ // guard below does not attempt to close the shared endpoint
WebSocket.
+ boolean alreadyOpen = getEndpoint().isWebSocketOpen();
AtomicReference<WebSocket> opened = new AtomicReference<>();
WebSocket webSocket = getEndpoint().getWebSocket(ws -> {
configureWebSocketHandlers(ws);
- opened.set(ws);
+ if (!alreadyOpen) {
+ opened.set(ws);
+ }
});
Review Comment:
This covers the start, but the stop half of CAMEL-25601 is still open.
`doStop` (line 101) only cancels the reconnect, so after a plain `stopRoute`
(without a restart) the old consumer's text/binary handlers stay on the
endpoint's WebSocket. They keep calling `handleResult` -> `processExchange` ->
`getProcessor().process(...)` for a consumer that is stopped, which is the
second half of the issue description.
Could `doStop` detach them under the same connect lock? For example, if the
endpoint's WebSocket is open, set `textMessageHandler(null)` and
`binaryMessageHandler(null)` (and `closeHandler(null)`). Two routes can consume
the same `consumeAsClient` URI and so share one endpoint. In that case it is
safest to detach only when this consumer registered the current handlers, for
example by recording the owning consumer on the endpoint when the setup
callback runs.
##########
components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumer.java:
##########
@@ -73,10 +73,16 @@ private WebSocket connect(boolean reconnect) throws
Exception {
Lock connectLock = getEndpoint().getClientConsumerConnectLock();
connectLock.lock();
try {
+ // Check before connecting: if the WebSocket is already open, the
setup callback will be invoked on it
+ // (see getWebSocket javadoc), but "opened" must only track
newly-opened sockets so that the reconnect
+ // guard below does not attempt to close the shared endpoint
WebSocket.
+ boolean alreadyOpen = getEndpoint().isWebSocketOpen();
Review Comment:
Minor, non-blocking: `alreadyOpen` is read before `getWebSocket` and outside
it. A producer on the same endpoint does not take the connect lock, so it can
open the socket between this check and `getWebSocket`. Then the callback runs
on the producer's socket with `alreadyOpen == false`, and a reconnect attempt
that finds the consumer stopped closes that shared socket. That needs a stop, a
reconnect and a producer send at the same moment, so it is very unlikely.
Letting `getWebSocket` tell the callback whether it opened the socket (it knows
which branch it took) would remove the window and the extra `isWebSocketOpen()`
helper.
##########
components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketClientConsumerHandlersTest.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.
+ * <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");
+
+ // Restart the consumer route. The WebSocket is already open — the fix
ensures handlers are re-registered.
+ context.getRouteController().startRoute("client-consumer");
Review Comment:
As far as I can tell, this scenario passes without the fix as well. A
restarted route gets a new consumer, but `DefaultRoute` passes it the same
route processor. Without the fix, the old (stopped) consumer's
`textMessageHandler` is still on the socket, and its `handleResult` ->
`getProcessor().process(...)` delivers `after-restart` to `mock:result` anyway.
So this test does not show that the new consumer registered its handlers. (The
producer-first test does, because there no handlers were registered at all.)
Once `doStop` detaches the handlers, this test becomes a real check.
Asserting that nothing arrives while the route is stopped would also cover the
stop half, for example `mock.expectedMessageCount(0);
mock.setAssertPeriod(500);` after `stopRoute`, then a
`template.sendBody(serverUri, ...)`.
--
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]
