This is an automated email from the ASF dual-hosted git repository.
gnodet pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new dff3af13d67c [backport camel-4.22.x] CAMEL-24789:
camel-vertx-websocket - finish an exchange with no peer synchronously (#26699)
dff3af13d67c is described below
commit dff3af13d67c13e41934a7540cfcda9f3401b675
Author: Guillaume Nodet <[email protected]>
AuthorDate: Mon Sep 21 21:52:49 2026 +0200
[backport camel-4.22.x] CAMEL-24789: camel-vertx-websocket - finish an
exchange with no peer synchronously (#26699)
---
.../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 dffba2fc51ba..1184a25f6bfb 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;
@@ -118,6 +119,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;
