You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2017/11/03 10:24:24 UTC
[camel] branch master updated: CAMEL-11989: camel-paho - Allow to
specify topic via header
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new 5a93fa5 CAMEL-11989: camel-paho - Allow to specify topic via header
5a93fa5 is described below
commit 5a93fa5cace03ec5c8990bfc76018dc2e4b1694b
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri Nov 3 11:24:12 2017 +0100
CAMEL-11989: camel-paho - Allow to specify topic via header
---
.../camel-paho/src/main/docs/paho-component.adoc | 9 +++
.../apache/camel/component/paho/PahoConstants.java | 2 +-
.../apache/camel/component/paho/PahoProducer.java | 11 ++-
.../component/paho/PahoOverrideTopicTest.java | 88 ++++++++++++++++++++++
4 files changed, 106 insertions(+), 4 deletions(-)
diff --git a/components/camel-paho/src/main/docs/paho-component.adoc b/components/camel-paho/src/main/docs/paho-component.adoc
index e07fef7..cc80c19 100644
--- a/components/camel-paho/src/main/docs/paho-component.adoc
+++ b/components/camel-paho/src/main/docs/paho-component.adoc
@@ -114,6 +114,7 @@ The following headers are recognized by the Paho component:
|Header |Java constant |Endpoint type |Value type |Description
|CamelMqttTopic |PahoConstants.MQTT_TOPIC |Consumer |String |The name of the topic
+|CamelPahoOverrideTopic |PahoConstants.CAMEL_PAHO_OVERRIDE_TOPIC |Producer |String |Name of topic to override and send to instead of topic specified on endpoint
|===
@@ -175,3 +176,11 @@ from("paho:some/queue?brokerUrl=tcp://iot.eclipse.org:1883")
.to("mock:test");
----
+And here we override the default topic and set to a dynamic topic
+
+[source,java]
+----
+from("direct:test")
+ .setHeader(PahoConstants.CAMEL_PAHO_OVERRIDE_TOPIC, simple("${header.customerId}"))
+ .to("paho:some/target/queue");
+----
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConstants.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConstants.java
index e7ecf90..598469e 100644
--- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConstants.java
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoConstants.java
@@ -33,7 +33,7 @@ public final class PahoConstants {
public static final String CAMEL_PAHO = "CamelPaho";
public static final String CAMEL_PAHO_MSG_QOS = CAMEL_PAHO + "Qos";
public static final String CAMEL_PAHO_MSG_RETAINED = CAMEL_PAHO + "Retained";
-
+ public static final String CAMEL_PAHO_OVERRIDE_TOPIC = CAMEL_PAHO + "OverrideTopic";
private PahoConstants() {
}
diff --git a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
index 777f52d..f99ff8f 100644
--- a/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
+++ b/components/camel-paho/src/main/java/org/apache/camel/component/paho/PahoProducer.java
@@ -20,9 +20,13 @@ import org.apache.camel.Exchange;
import org.apache.camel.impl.DefaultProducer;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class PahoProducer extends DefaultProducer {
+ private static final Logger LOG = LoggerFactory.getLogger(PahoProducer.class);
+
public PahoProducer(PahoEndpoint endpoint) {
super(endpoint);
}
@@ -30,16 +34,17 @@ public class PahoProducer extends DefaultProducer {
@Override
public void process(Exchange exchange) throws Exception {
MqttClient client = getEndpoint().getClient();
- String topic = getEndpoint().getTopic();
-
+
+ String topic = exchange.getIn().getHeader(PahoConstants.CAMEL_PAHO_OVERRIDE_TOPIC, getEndpoint().getTopic(), String.class);
int qos = exchange.getIn().getHeader(PahoConstants.CAMEL_PAHO_MSG_QOS, getEndpoint().getQos(), Integer.class);
boolean retained = exchange.getIn().getHeader(PahoConstants.CAMEL_PAHO_MSG_RETAINED, getEndpoint().isRetained(), Boolean.class);
-
byte[] payload = exchange.getIn().getBody(byte[].class);
MqttMessage message = new MqttMessage(payload);
message.setQos(qos);
message.setRetained(retained);
+
+ LOG.debug("Publishing to topic: {}, qos: {}, retrained: {}", topic, qos, retained);
client.publish(topic, message);
}
diff --git a/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoOverrideTopicTest.java b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoOverrideTopicTest.java
new file mode 100644
index 0000000..0470b13
--- /dev/null
+++ b/components/camel-paho/src/test/java/org/apache/camel/component/paho/PahoOverrideTopicTest.java
@@ -0,0 +1,88 @@
+/**
+ * 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 org.apache.activemq.broker.BrokerService;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.JndiRegistry;
+import org.apache.camel.test.AvailablePortFinder;
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.junit.Test;
+
+public class PahoOverrideTopicTest extends CamelTestSupport {
+
+ MqttConnectOptions connectOptions = new MqttConnectOptions();
+
+ BrokerService broker;
+
+ int mqttPort = AvailablePortFinder.getNextAvailable();
+
+ @Override
+ protected boolean useJmx() {
+ return false;
+ }
+
+ @Override
+ public void doPreSetup() throws Exception {
+ super.doPreSetup();
+ broker = new BrokerService();
+ broker.setPersistent(false);
+ broker.addConnector("mqtt://localhost:" + mqttPort);
+ broker.start();
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ broker.stop();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:test").to("paho:queue?brokerUrl=tcp://localhost:" + mqttPort);
+ from("paho:myoverride?brokerUrl=tcp://localhost:" + mqttPort).to("mock:test");
+
+ }
+ };
+ }
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry registry = super.createRegistry();
+ registry.bind("connectOptions", connectOptions);
+ return registry;
+ }
+
+ // Tests
+
+ @Test
+ public void shouldOverride() throws InterruptedException {
+ // Given
+ getMockEndpoint("mock:test").expectedMessageCount(1);
+
+ // When
+ template.sendBodyAndHeader("direct:test", "Hello World", PahoConstants.CAMEL_PAHO_OVERRIDE_TOPIC, "myoverride");
+
+ // Then
+ assertMockEndpointsSatisfied();
+ }
+
+}
--
To stop receiving notification emails like this one, please contact
['"commits@camel.apache.org" <co...@camel.apache.org>'].