This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new aaef7487c407 CAMEL-24789: camel-vertx-websocket - finish an exchange
with no peer synchronously (#26541)
aaef7487c407 is described below
commit aaef7487c407c6f66f2a4013c224105119a18bc0
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Sep 21 14:47:01 2026 +0200
CAMEL-24789: camel-vertx-websocket - finish an exchange with no peer
synchronously (#26541)
* CAMEL-24789: camel-vertx-websocket - finish an exchange with no peer
synchronously
With no peer to send to, the producer completed its callback with
doneSync=true
and then returned false, which tells the caller the opposite: that the
exchange
will be finished later, from a write handler that never runs because there
is
nothing to write to. AsyncProcessor#process documents the return value as
doneSync, and the null-body branch a few lines earlier in the same method
already
gets this right.
The empty case now completes and returns true, and says at WARN that the
message
was not delivered. The path is easy to reach: sendToAll with nothing
connected
yet, or a CamelVertxWebsocket.connectionKey that matches no current peer.
A connection key that matches nothing was also dropped in silence, while a
peer
whose websocket turned out to be null one line further on did warn. Both
report
now.
Co-Authored-By: Claude Opus 5 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
* CAMEL-24789: write the test in the assertion style of the module
camel-vertx-websocket is written against org.junit.jupiter.api.Assertions
in all
of its other test classes, so the new one follows suit rather than arriving
as
the only AssertJ file, which also takes the assertj test dependency back
out.
Same point davsclaus raised on CAMEL-24788.
Co-Authored-By: Claude Opus 5 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
* CAMEL-24789: camel-vertx-websocket - cover the two gaps left in the
no-peer tests
anExchangeWithNoBodyIsDoneSynchronously asserted only that process()
returned
true and that doneSync was true, and a callback completed twice satisfies
both.
It now counts callbacks like its sibling does, so a double-invocation fails.
The unmatched connection-key path had no test at all. It is only reachable
when
peers exist and the key matches none of them, which needs a real consumer -
peers are added from a live connection handler, so the producer unit test
cannot get there. The test therefore lives in VertxWebsocketTest, where the
server harness already is: one connected peer, a key nobody has, and an
assertion that the peer receives nothing.
That pins behaviour rather than the log line, which would need a log
appender
to assert and would be brittle. It is the behaviour that matters: mutating
the
branch to fall back to putAll(peers) - the plausible wrong fix - makes the
test
fail, so an unmatched key silently fanning out to whoever is connected is
now
caught.
Co-Authored-By: Claude Opus 5 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
* CAMEL-24789: log a broadcast that reaches nobody at debug
Having no peer connected is an ordinary state for a broadcast - a route
pushing
status on a timer before any browser attaches hits it on every tick - so
warning
each time turns a normal condition into log noise. That line drops to debug.
The misconfiguration keeps its warning: a connection key that matches no
peer is
still reported at WARN from getConnectedPeers, which is the case where the
route
asked for a specific peer and did not get it.
Co-Authored-By: Claude Opus 5 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---------
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 <[email protected]>
---
.../vertx/websocket/VertxWebsocketProducer.java | 21 ++++-
.../VertxWebsocketProducerNoPeerTest.java | 101 +++++++++++++++++++++
.../vertx/websocket/VertxWebsocketTest.java | 29 ++++++
3 files changed, 147 insertions(+), 4 deletions(-)
diff --git
a/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketProducer.java
b/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketProducer.java
index a6680c5ed8ed..28ab393b84a3 100644
---
a/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketProducer.java
+++
b/components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketProducer.java
@@ -60,13 +60,20 @@ public class VertxWebsocketProducer extends
DefaultAsyncProducer {
}
Map<String, WebSocketBase> connectedPeers =
getConnectedPeers(exchange);
- VertxWebsocketResultHandler vertxWebsocketResultHandler
- = new VertxWebsocketResultHandler(exchange, callback,
connectedPeers.keySet());
if (connectedPeers.isEmpty()) {
+ // nothing was sent, so the exchange is done here rather than
from a write handler. Having nobody
+ // connected is an ordinary state for a broadcast, so it is
only worth a debug line: a connection
+ // key that matches no peer is the misconfiguration, and
getConnectedPeers warns about that one
+ LOG.debug("No WebSocket peer to send to for endpoint {}, the
message is not delivered",
+ getEndpoint().getEndpointUri());
callback.done(true);
+ return true;
}
+ VertxWebsocketResultHandler vertxWebsocketResultHandler
+ = new VertxWebsocketResultHandler(exchange, callback,
connectedPeers.keySet());
+
// Send message to each peer then record and process the results
asynchronously
connectedPeers.forEach((connectionKey, webSocket) -> {
Handler<AsyncResult<Void>> handler = result -> {
@@ -121,8 +128,14 @@ public class VertxWebsocketProducer extends
DefaultAsyncProducer {
String connectionKey =
message.getHeader(VertxWebsocketConstants.CONNECTION_KEY, String.class);
if (connectionKey != null && ObjectHelper.isNotEmpty(peers)) {
Stream.of(connectionKey.split(","))
- .filter(peers::containsKey)
- .forEach(key -> connectedPeers.put(key,
endpoint.findPeerForConnectionKey(key)));
+ .forEach(key -> {
+ if (peers.containsKey(key)) {
+ connectedPeers.put(key,
endpoint.findPeerForConnectionKey(key));
+ } else {
+ // a key that matches nothing would otherwise
be dropped without a word
+ LOG.warn("No WebSocket peer connection found
for connection key {}", key);
+ }
+ });
} else {
// The producer is invoking an external server not managed by
camel
connectedPeers.put(UUID.randomUUID().toString(),
endpoint.getWebSocket(exchange));
diff --git
a/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketProducerNoPeerTest.java
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketProducerNoPeerTest.java
new file mode 100644
index 000000000000..136017a60c31
--- /dev/null
+++
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketProducerNoPeerTest.java
@@ -0,0 +1,101 @@
+/*
+ * 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.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * An exchange with no peer to send to is finished by the producer itself, so
it has to report that it was done
+ * synchronously. Builds the endpoint directly, so nothing reaches a server.
+ */
+class VertxWebsocketProducerNoPeerTest {
+
+ private DefaultCamelContext context;
+
+ @AfterEach
+ void tearDown() {
+ if (context != null) {
+ context.stop();
+ }
+ }
+
+ private VertxWebsocketProducer producer() throws Exception {
+ context = new DefaultCamelContext();
+
+ VertxWebsocketComponent component = new VertxWebsocketComponent();
+ component.setCamelContext(context);
+
+ VertxWebsocketEndpoint endpoint
+ = (VertxWebsocketEndpoint)
component.createEndpoint("vertx-websocket:localhost:1234/test");
+ return (VertxWebsocketProducer) endpoint.createProducer();
+ }
+
+ @Test
+ void anExchangeWithNoPeerIsDoneSynchronously() throws Exception {
+ VertxWebsocketProducer producer = producer();
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody("a message nobody is listening for");
+ // broadcasting to an empty host registry is the one path that reaches
no peer without opening a connection
+ exchange.getIn().setHeader(VertxWebsocketConstants.SEND_TO_ALL, true);
+
+ AtomicInteger callbacks = new AtomicInteger();
+ AtomicReference<Boolean> doneSync = new AtomicReference<>();
+
+ // the callback used to be completed with doneSync=true while the
method returned false, which says the
+ // opposite: that the exchange would be finished from a write handler
that never runs
+ boolean result = producer.process(exchange, sync -> {
+ callbacks.incrementAndGet();
+ doneSync.set(sync);
+ });
+
+ assertTrue(result, "process must report that it finished the exchange
itself");
+ assertEquals(1, callbacks.get());
+ assertTrue(doneSync.get());
+ assertNull(exchange.getException());
+ }
+
+ @Test
+ void anExchangeWithNoBodyIsDoneSynchronously() throws Exception {
+ VertxWebsocketProducer producer = producer();
+
+ Exchange exchange = new DefaultExchange(context);
+
+ AtomicInteger callbacks = new AtomicInteger();
+ AtomicReference<Boolean> doneSync = new AtomicReference<>();
+ boolean result = producer.process(exchange, sync -> {
+ callbacks.incrementAndGet();
+ doneSync.set(sync);
+ });
+
+ assertTrue(result, "process must report that it finished the exchange
itself");
+ // the same triple as the first test: completing the callback twice
would also satisfy the other two
+ assertEquals(1, callbacks.get());
+ assertTrue(doneSync.get());
+ }
+}
diff --git
a/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketTest.java
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketTest.java
index 03e006752dff..0a336cb58291 100644
---
a/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketTest.java
+++
b/components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketTest.java
@@ -38,6 +38,7 @@ import org.apache.camel.component.mock.MockEndpoint;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -117,6 +118,34 @@ public class VertxWebsocketTest extends
VertxWebSocketTestSupport {
assertTrue(results.contains("Hello World"));
}
+ @Test
+ void sendWithAnUnmatchedConnectionKeyDeliversToNobody() throws Exception {
+ // the branch this covers needs peers to exist while the key matches
none of them: a key that is simply
+ // absent from a populated registry is dropped, and before CAMEL-24789
it was dropped without a word
+ CountDownLatch connected = new CountDownLatch(1);
+ List<String> results = new ArrayList<>();
+ openWebSocketConnection("localhost", port.getPort(), "/test", message
-> {
+ synchronized (results) {
+ results.add(message);
+ connected.countDown();
+ }
+ });
+
+ VertxWebsocketEndpoint endpoint
+ = context.getEndpoint("vertx-websocket:localhost:" + port +
"/test", VertxWebsocketEndpoint.class);
+ awaitConnectedPeers(endpoint, 1);
+
+ template.sendBodyAndHeader("vertx-websocket:localhost:" + port +
"/test", "Hello World",
+ VertxWebsocketConstants.CONNECTION_KEY,
"a-key-no-peer-ever-had");
+
+ // the send returns rather than hanging, and the connected peer is
left untouched - the message went
+ // nowhere, which is the point: an unmatched key must not silently fan
out to whoever happens to be there
+ assertFalse(connected.await(2, TimeUnit.SECONDS), "no peer should have
received the message");
+ synchronized (results) {
+ assertEquals(List.of(), results);
+ }
+ }
+
@Test
void testSendWithConnectionKeyForParameterizedPath() throws Exception {
int expectedResultCount = 1;
