This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 faa015319ee7 CAMEL-25853: camel-paho, camel-paho-mqtt5, camel-hivemq -
toD publishes to its own topic (#28326)
faa015319ee7 is described below
commit faa015319ee73edf6f93fd8351e70e49b5a2b5c7
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 23 13:52:29 2026 +0530
CAMEL-25853: camel-paho, camel-paho-mqtt5, camel-hivemq - toD publishes to
its own topic (#28326)
The toD optimisation left the override topic header on the message, so a
second toD of the same exchange published to the topic of the first one. A post
processor now removes the header that the toD set.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/hivemq/HiveMQSendDynamicAware.java | 14 +-
.../camel/component/hivemq/HiveMQToDTwiceTest.java | 143 +++++++++++++++++++++
.../paho/mqtt5/PahoMqtt5SendDynamicAware.java | 15 ++-
.../paho/mqtt5/PahoMqtt5ToDTwiceTest.java | 93 ++++++++++++++
.../camel/component/paho/PahoSendDynamicAware.java | 15 ++-
.../camel/component/paho/PahoToDTwiceTest.java | 93 ++++++++++++++
6 files changed, 368 insertions(+), 5 deletions(-)
diff --git
a/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQSendDynamicAware.java
b/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQSendDynamicAware.java
index c5ae3565750a..e4fb9914fafc 100644
---
a/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQSendDynamicAware.java
+++
b/components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQSendDynamicAware.java
@@ -71,7 +71,19 @@ public class HiveMQSendDynamicAware extends
SendDynamicAwareSupport {
@Override
public Processor createPostProcessor(Exchange exchange, DynamicAwareEntry
entry) throws Exception {
- return null;
+ if (exchange.getMessage().getHeader(HiveMQConstants.OVERRIDE_TOPIC) !=
null) {
+ // the topic header was set before toD, so it is not ours to remove
+ return null;
+ }
+
+ // remove the topic header the pre processor has set, so a later send
of the same exchange
+ // (another toD, a loop) is not published to this topic
+ return new Processor() {
+ @Override
+ public void process(Exchange exchange) {
+
exchange.getMessage().removeHeader(HiveMQConstants.OVERRIDE_TOPIC);
+ }
+ };
}
private String parseTopicName(String uri) {
diff --git
a/components/camel-hivemq/src/test/java/org/apache/camel/component/hivemq/HiveMQToDTwiceTest.java
b/components/camel-hivemq/src/test/java/org/apache/camel/component/hivemq/HiveMQToDTwiceTest.java
new file mode 100644
index 000000000000..4be2d6ce70e9
--- /dev/null
+++
b/components/camel-hivemq/src/test/java/org/apache/camel/component/hivemq/HiveMQToDTwiceTest.java
@@ -0,0 +1,143 @@
+/*
+ * 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.hivemq;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.Consumer;
+
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Two toD to hivemq in the same exchange must publish to two topics: the
topic header that the toD optimisation sets
+ * for the first send must not be used by the second send.
+ */
+class HiveMQToDTwiceTest {
+
+ private final List<String> published = new CopyOnWriteArrayList<>();
+
+ @Test
+ void secondToDPublishesToItsOwnTopic() throws Exception {
+ try (CamelContext context = createContext()) {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .toD("hivemq:${header.first}")
+ .toD("hivemq:${header.second}");
+ }
+ });
+ context.start();
+
+
context.createProducerTemplate().sendBodyAndHeaders("direct:start", "Hello",
+ Map.of("first", "status/1", "second", "audit/1"));
+
+ assertThat(published).containsExactly("status/1", "audit/1");
+ }
+ }
+
+ @Test
+ void loopWithToDPublishesEachIterationToItsOwnTopic() throws Exception {
+ try (CamelContext context = createContext()) {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:loop")
+ .loop(3)
+
.toD("hivemq:devices/${exchangeProperty.CamelLoopIndex}/cmd")
+ .end();
+ }
+ });
+ context.start();
+
+ context.createProducerTemplate().sendBody("direct:loop", "Hello");
+
+ assertThat(published).containsExactly("devices/0/cmd",
"devices/1/cmd", "devices/2/cmd");
+ }
+ }
+
+ private CamelContext createContext() {
+ CamelContext context = new DefaultCamelContext();
+ context.addComponent("hivemq", new HiveMQComponent() {
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters)
+ throws Exception {
+ HiveMQEndpoint endpoint = new HiveMQEndpoint(uri, this,
getConfiguration().copy(), remaining) {
+ @Override
+ HiveMQClientAdapter createClient() {
+ return new RecordingClient();
+ }
+ };
+ setProperties(endpoint, parameters);
+ return endpoint;
+ }
+ });
+ return context;
+ }
+
+ private final class RecordingClient implements HiveMQClientAdapter {
+ @Override
+ public CompletableFuture<?> connect(boolean cleanStart) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public void stop() {
+ }
+
+ @Override
+ public boolean isConnected() {
+ return true;
+ }
+
+ @Override
+ public boolean isConnectedOrReconnecting() {
+ return true;
+ }
+
+ @Override
+ public CompletableFuture<?> subscribe(String topicFilter, MqttQos qos,
Consumer<HiveMQMessage> callback) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<?> unsubscribe(String topicFilter) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<?> publish(String topic, byte[] payload,
MqttQos qos, boolean retained) {
+ published.add(topic);
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public <T> Optional<T> getClient(Class<T> clazz) {
+ return Optional.empty();
+ }
+ }
+}
diff --git
a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5SendDynamicAware.java
b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5SendDynamicAware.java
index 57b66644b526..ce28f16bcb12 100644
---
a/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5SendDynamicAware.java
+++
b/components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5SendDynamicAware.java
@@ -98,8 +98,19 @@ public class PahoMqtt5SendDynamicAware extends
ServiceSupport implements SendDyn
@Override
public Processor createPostProcessor(Exchange exchange, DynamicAwareEntry
entry) throws Exception {
- // no post processor is needed
- return null;
+ if
(exchange.getMessage().getHeader(PahoMqtt5Constants.CAMEL_PAHO_OVERRIDE_TOPIC)
!= null) {
+ // the topic header was set before toD, so it is not ours to remove
+ return null;
+ }
+
+ // remove the topic header the pre processor has set, so a later send
of the same exchange
+ // (another toD, a loop) is not published to this topic
+ return new Processor() {
+ @Override
+ public void process(Exchange exchange) {
+
exchange.getMessage().removeHeader(PahoMqtt5Constants.CAMEL_PAHO_OVERRIDE_TOPIC);
+ }
+ };
}
private String parseTopicName(String uri) {
diff --git
a/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ToDTwiceTest.java
b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ToDTwiceTest.java
new file mode 100644
index 000000000000..08f63405133c
--- /dev/null
+++
b/components/camel-paho-mqtt5/src/test/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5ToDTwiceTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.paho.mqtt5;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.eclipse.paho.mqttv5.client.MqttClient;
+import org.eclipse.paho.mqttv5.common.MqttMessage;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Two toD to paho-mqtt5 in the same exchange must publish to two topics: the
topic header that the toD optimisation
+ * sets for the first send must not be used by the second send.
+ */
+class PahoMqtt5ToDTwiceTest {
+
+ @Test
+ void secondToDPublishesToItsOwnTopic() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ try (CamelContext context = new DefaultCamelContext()) {
+ PahoMqtt5Component paho = new PahoMqtt5Component();
+ paho.setClient(client);
+ context.addComponent("paho-mqtt5", paho);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .toD("paho-mqtt5:${header.first}")
+ .toD("paho-mqtt5:${header.second}");
+ }
+ });
+ context.start();
+
+
context.createProducerTemplate().sendBodyAndHeaders("direct:start", "Hello",
+ Map.of("first", "status/1", "second", "audit/1"));
+
+ ArgumentCaptor<String> topics =
ArgumentCaptor.forClass(String.class);
+ verify(client, times(2)).publish(topics.capture(),
any(MqttMessage.class));
+ assertEquals(List.of("status/1", "audit/1"),
topics.getAllValues());
+ }
+ }
+
+ @Test
+ void loopWithToDPublishesEachIterationToItsOwnTopic() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ try (CamelContext context = new DefaultCamelContext()) {
+ PahoMqtt5Component paho = new PahoMqtt5Component();
+ paho.setClient(client);
+ context.addComponent("paho-mqtt5", paho);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:loop")
+ .loop(3)
+
.toD("paho-mqtt5:devices/${exchangeProperty.CamelLoopIndex}/cmd")
+ .end();
+ }
+ });
+ context.start();
+
+ context.createProducerTemplate().sendBody("direct:loop", "Hello");
+
+ ArgumentCaptor<String> topics =
ArgumentCaptor.forClass(String.class);
+ verify(client, times(3)).publish(topics.capture(),
any(MqttMessage.class));
+ assertEquals(List.of("devices/0/cmd", "devices/1/cmd",
"devices/2/cmd"), topics.getAllValues());
+ }
+ }
+}
diff --git
a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoSendDynamicAware.java
b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoSendDynamicAware.java
index 01c59d06a06b..075e0f214d19 100644
---
a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoSendDynamicAware.java
+++
b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoSendDynamicAware.java
@@ -99,8 +99,19 @@ public class PahoSendDynamicAware extends ServiceSupport
implements SendDynamicA
@Override
public Processor createPostProcessor(Exchange exchange, DynamicAwareEntry
entry) throws Exception {
- // no post processor is needed
- return null;
+ if
(exchange.getMessage().getHeader(PahoConstants.CAMEL_PAHO_OVERRIDE_TOPIC) !=
null) {
+ // the topic header was set before toD, so it is not ours to remove
+ return null;
+ }
+
+ // remove the topic header the pre processor has set, so a later send
of the same exchange
+ // (another toD, a loop) is not published to this topic
+ return new Processor() {
+ @Override
+ public void process(Exchange exchange) {
+
exchange.getMessage().removeHeader(PahoConstants.CAMEL_PAHO_OVERRIDE_TOPIC);
+ }
+ };
}
private String parseTopicName(String uri) {
diff --git
a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoToDTwiceTest.java
b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoToDTwiceTest.java
new file mode 100644
index 000000000000..6c26953e6efb
--- /dev/null
+++
b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoToDTwiceTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.paho;
+
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+/**
+ * Two toD to paho in the same exchange must publish to two topics: the topic
header that the toD optimisation sets for
+ * the first send must not be used by the second send.
+ */
+class PahoToDTwiceTest {
+
+ @Test
+ void secondToDPublishesToItsOwnTopic() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ try (CamelContext context = new DefaultCamelContext()) {
+ PahoComponent paho = new PahoComponent();
+ paho.setClient(client);
+ context.addComponent("paho", paho);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .toD("paho:${header.first}")
+ .toD("paho:${header.second}");
+ }
+ });
+ context.start();
+
+
context.createProducerTemplate().sendBodyAndHeaders("direct:start", "Hello",
+ Map.of("first", "status/1", "second", "audit/1"));
+
+ ArgumentCaptor<String> topics =
ArgumentCaptor.forClass(String.class);
+ verify(client, times(2)).publish(topics.capture(),
any(MqttMessage.class));
+ assertEquals(List.of("status/1", "audit/1"),
topics.getAllValues());
+ }
+ }
+
+ @Test
+ void loopWithToDPublishesEachIterationToItsOwnTopic() throws Exception {
+ MqttClient client = mock(MqttClient.class);
+ try (CamelContext context = new DefaultCamelContext()) {
+ PahoComponent paho = new PahoComponent();
+ paho.setClient(client);
+ context.addComponent("paho", paho);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:loop")
+ .loop(3)
+
.toD("paho:devices/${exchangeProperty.CamelLoopIndex}/cmd")
+ .end();
+ }
+ });
+ context.start();
+
+ context.createProducerTemplate().sendBody("direct:loop", "Hello");
+
+ ArgumentCaptor<String> topics =
ArgumentCaptor.forClass(String.class);
+ verify(client, times(3)).publish(topics.capture(),
any(MqttMessage.class));
+ assertEquals(List.of("devices/0/cmd", "devices/1/cmd",
"devices/2/cmd"), topics.getAllValues());
+ }
+ }
+}
