gnodet-bot commented on code in PR #26533:
URL: https://github.com/apache/camel/pull/26533#discussion_r4034461152
##########
components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketHost.java:
##########
@@ -162,7 +163,9 @@ public void connect(VertxWebsocketConsumer consumer) {
public void disconnect(String path) {
LOG.info("Disconnected consumer for path {}", path);
Route route = routeRegistry.remove(path);
- route.remove();
+ if (route != null) {
+ route.remove();
+ }
if (routeRegistry.isEmpty()) {
Review Comment:
⚠️ **Non-atomic check-then-act: double `stop()` is possible under concurrent
disconnects.**
`routeRegistry` is now a `ConcurrentHashMap`, so `.remove()` and
`.isEmpty()` are each individually atomic — but the compound sequence
`remove(path) → isEmpty() → stop()` is not. The commit message states the route
controller stops consumers *possibly concurrently*, so this race is reachable:
1. T1: `routeRegistry.remove("/a")` → `{"/b"}` remains
2. T2: `routeRegistry.remove("/b")` → `{}` empty
3. T2: `isEmpty()` = true → calls `stop()`
4. T1: `isEmpty()` = true → calls `stop()` again (server is already closing)
`stop()` is not synchronized and `server` is not `volatile`, so both threads
can see `server != null` and both call `server.close()` on the same
`HttpServer` object concurrently.
Fix: synchronize the empty-check/stop pair — lifecycle operations, not hot
paths:
##########
components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketHost.java:
##########
@@ -236,6 +239,14 @@ public void stop() throws ExecutionException,
InterruptedException {
port = VertxWebsocketConstants.DEFAULT_VERTX_SERVER_PORT;
}
+ /**
+ * Whether this host still serves any consumer. Every consumer bound to
the same host and port shares one instance,
+ * so the host outlives the first consumer that stops, and only once the
last one goes is its server stopped.
+ */
+ public boolean isServingConsumers() {
Review Comment:
⚠️ **`server` field is not `volatile` — data race between `start()` and
`stop()`.**
`start()` checks `if (server == null)` and `stop()` checks `if (server !=
null)` then sets `this.server = null` in a `finally` block. None of these
accesses are on a `volatile` field or inside a synchronized block. In the
concurrent-lifecycle scenario the commit describes, there is no happens-before
edge between `stop()` writing `null` and a subsequent `start()` reading it.
Minimal fix: declare `private volatile HttpServer server;` at line 58.
##########
components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketMultiConsumerLifecycleTest.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.util.Map;
+
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Consumers bound to the same host and port share one {@link
VertxWebsocketHost}, so stopping one of them must leave
+ * the others able to serve and, later, to disconnect.
+ */
+public class VertxWebsocketMultiConsumerLifecycleTest extends
VertxWebSocketTestSupport {
+
+ private Map<VertxWebsocketHostKey, VertxWebsocketHost> hostRegistry() {
+ return context.getComponent("vertx-websocket",
VertxWebsocketComponent.class).getVertxHostRegistry();
+ }
+
+ @Test
+ void theHostOutlivesTheFirstConsumerToStop() throws Exception {
+ assertThat(hostRegistry()).hasSize(1);
+
+ context.getRouteController().stopRoute("a");
+
+ // the host still serves route b, so it must still be reachable - it
used to be dropped from the
+ // registry here, which left b unable to ever disconnect and its
server running for good
+ assertThat(hostRegistry()).hasSize(1);
+
+ MockEndpoint mockEndpoint = getMockEndpoint("mock:result");
+ mockEndpoint.expectedBodiesReceived("Hello b");
+
+ template.sendBody("vertx-websocket:localhost:" + port + "/test/b",
"b");
+
+ mockEndpoint.assertIsSatisfied();
+ }
+
+ @Test
+ void theHostGoesWithTheLastConsumerToStop() throws Exception {
+ assertThat(hostRegistry()).hasSize(1);
+
+ context.getRouteController().stopRoute("a");
+ context.getRouteController().stopRoute("b");
+
+ assertThat(hostRegistry()).isEmpty();
+ }
+
+ @Test
+ void stoppingAConsumerTwiceIsHarmless() throws Exception {
Review Comment:
💡 **Test gap: the concurrent-stop scenario that motivates
`ConcurrentHashMap` is not exercised.**
`stoppingAConsumerTwiceIsHarmless` is sequential — it verifies idempotence
of a single-threaded stop/start/stop cycle, not thread safety. The race that
justifies switching from `HashMap` to `ConcurrentHashMap` — two consumers on
the same host stopped simultaneously — has no test. Add a test that stops
routes `"a"` and `"b"` from two concurrent threads and asserts the host
registry is empty without exceptions.
--
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]
