CAMEL-24789: camel-vertx-websocket - finish an exchange with no peer synchronously

components/camel-vertx/camel-vertx-websocket/src/main/java/org/apache/camel/component/vertx/websocket/VertxWebsocketProducer.java

@@ -60,13 +60,20 @@ public boolean process(Exchange exchange, AsyncCallback callback) {
             }
 
             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 @@ private Map<String, WebSocketBase> getConnectedPeers(Exchange exchange) throws E
             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));

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());
+    }
+}

components/camel-vertx/camel-vertx-websocket/src/test/java/org/apache/camel/component/vertx/websocket/VertxWebsocketTest.java

@@ -38,6 +38,7 @@
 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 void testSendWithConnectionKey() throws Exception {
         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;