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 2012/10/06 17:10:29 UTC
svn commit: r1395089 - in /camel/branches/camel-2.10.x: ./
components/camel-mqtt/
components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/
components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/
components/camel-mqtt/src/test/r...
Author: davsclaus
Date: Sat Oct 6 15:10:28 2012
New Revision: 1395089
URL: http://svn.apache.org/viewvc?rev=1395089&view=rev
Log:
CAMEL-5670: Fixed NPE in MQTT publish. Refactored code to leverage async routing engine. Polished code.
Removed:
camel/branches/camel-2.10.x/components/camel-mqtt/src/test/resources/test-options.properties
Modified:
camel/branches/camel-2.10.x/ (props changed)
camel/branches/camel-2.10.x/components/camel-mqtt/ (props changed)
camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTComponent.java
camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConfiguration.java
camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConsumer.java
camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTProducer.java
camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTConfigurationTest.java
camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTConsumerTest.java
camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTProducerTest.java
camel/branches/camel-2.10.x/components/camel-mqtt/src/test/resources/log4j.properties
Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
Merged /camel/trunk:r1395085
Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.
Propchange: camel/branches/camel-2.10.x/components/camel-mqtt/
------------------------------------------------------------------------------
--- svn:ignore (original)
+++ svn:ignore Sat Oct 6 15:10:28 2012
@@ -6,3 +6,4 @@ target
camel-mqtt.iml
camel-mqtt.ipr
camel-mqtt.iws
+*.idea
Modified: camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTComponent.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTComponent.java?rev=1395089&r1=1395088&r2=1395089&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTComponent.java (original)
+++ camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTComponent.java Sat Oct 6 15:10:28 2012
@@ -22,7 +22,7 @@ import org.apache.camel.Endpoint;
import org.apache.camel.impl.DefaultComponent;
/**
- * MQTT component
+ * MQTT Component
*/
public class MQTTComponent extends DefaultComponent {
Modified: camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConfiguration.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConfiguration.java?rev=1395089&r1=1395088&r2=1395089&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConfiguration.java (original)
+++ camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConfiguration.java Sat Oct 6 15:10:28 2012
@@ -20,6 +20,7 @@ import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;
public class MQTTConfiguration extends MQTT {
+
/**
* These a properties that are looked for in an Exchange - to publish to
*/
@@ -28,7 +29,7 @@ public class MQTTConfiguration extends M
private String mqttQosPropertyName = "MQTTQos";
/**
- * These are set on the Endpoint - together with properties inheritted from MQTT
+ * These are set on the Endpoint - together with properties inherited from MQTT
*/
private String subscribeTopicName = "";
private String publishTopicName = "camel/mqtt/test";
@@ -40,7 +41,6 @@ public class MQTTConfiguration extends M
private boolean byDefaultRetain;
private QoS qos = QoS.AT_LEAST_ONCE;
-
public String getQualityOfService() {
return qualityOfService;
}
@@ -94,7 +94,6 @@ public class MQTTConfiguration extends M
this.mqttQosPropertyName = mqttQosPropertyName;
}
-
public int getConnectWaitInSeconds() {
return connectWaitInSeconds;
}
@@ -119,7 +118,6 @@ public class MQTTConfiguration extends M
this.sendWaitInSeconds = sendWaitInSeconds;
}
-
public boolean isByDefaultRetain() {
return byDefaultRetain;
}
@@ -128,9 +126,7 @@ public class MQTTConfiguration extends M
this.byDefaultRetain = byDefaultRetain;
}
-
static QoS getQoS(String qualityOfService) {
-
for (QoS q : QoS.values()) {
if (q.name().equalsIgnoreCase(qualityOfService)) {
return q;
@@ -147,6 +143,7 @@ public class MQTTConfiguration extends M
}
throw new IllegalArgumentException("There is no QoS with name " + qualityOfService);
}
+
}
Modified: camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConsumer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConsumer.java?rev=1395089&r1=1395088&r2=1395089&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConsumer.java (original)
+++ camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTConsumer.java Sat Oct 6 15:10:28 2012
@@ -21,21 +21,35 @@ import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultConsumer;
public class MQTTConsumer extends DefaultConsumer {
+
public MQTTConsumer(MQTTEndpoint endpoint, Processor processor) {
super(endpoint, processor);
}
+ @Override
+ public MQTTEndpoint getEndpoint() {
+ return (MQTTEndpoint) super.getEndpoint();
+ }
+
protected void doStart() throws Exception {
- ((MQTTEndpoint) getEndpoint()).addConsumer(this);
+ getEndpoint().addConsumer(this);
super.doStart();
}
protected void doStop() throws Exception {
- ((MQTTEndpoint) getEndpoint()).removeConsumer(this);
+ getEndpoint().removeConsumer(this);
super.doStop();
}
- void processExchange(Exchange exchange) throws Exception {
- getProcessor().process(exchange);
+ void processExchange(Exchange exchange) {
+ try {
+ getProcessor().process(exchange);
+ } catch (Throwable e) {
+ exchange.setException(e);
+ }
+
+ if (exchange.getException() != null) {
+ getExceptionHandler().handleException("Error processing exchange.", exchange, exchange.getException());
+ }
}
}
Modified: camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java?rev=1395089&r1=1395088&r2=1395089&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java (original)
+++ camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java Sat Oct 6 15:10:28 2012
@@ -53,14 +53,12 @@ public class MQTTEndpoint extends Defaul
@Override
public Consumer createConsumer(Processor processor) throws Exception {
- MQTTConsumer consumer = new MQTTConsumer(this, processor);
- return consumer;
+ return new MQTTConsumer(this, processor);
}
@Override
public Producer createProducer() throws Exception {
- MQTTProducer producer = new MQTTProducer(this);
- return producer;
+ return new MQTTProducer(this);
}
public MQTTConfiguration getConfiguration() {
@@ -74,31 +72,25 @@ public class MQTTEndpoint extends Defaul
connection.listener(new Listener() {
public void onConnected() {
- LOG.info("MQTT Endpoint Connected to " + configuration.getHost());
+ LOG.info("MQTT Connection connected to {}", configuration.getHost());
}
public void onDisconnected() {
- LOG.debug("MQTT Connection disconnected");
+ LOG.debug("MQTT Connection disconnected from {}", configuration.getHost());
}
public void onPublish(UTF8Buffer topic, Buffer body, Runnable ack) {
-
if (!consumers.isEmpty()) {
Exchange exchange = createExchange();
exchange.getIn().setBody(body.toByteArray());
exchange.setProperty(configuration.getMqttTopicPropertyName(), topic.toString());
for (MQTTConsumer consumer : consumers) {
- try {
- consumer.processExchange(exchange);
- } catch (Exception e) {
- LOG.error("Failed to process exchange ", exchange);
- }
+ consumer.processExchange(exchange);
}
}
if (ack != null) {
ack.run();
}
-
}
public void onFailure(Throwable value) {
@@ -106,12 +98,13 @@ public class MQTTEndpoint extends Defaul
public void onSuccess(Void value) {
}
- public void onFailure(Throwable value) {
- LOG.debug("Failed to disconnect from " + configuration.getHost());
+ public void onFailure(Throwable e) {
+ LOG.debug("Failed to disconnect from " + configuration.getHost() + ". This exception is ignored.", e);
}
});
}
});
+
final Promise<Object> promise = new Promise<Object>();
connection.connect(new Callback<Void>() {
public void onSuccess(Void value) {
@@ -161,8 +154,8 @@ public class MQTTEndpoint extends Defaul
super.doStop();
}
- void publish(String topic, byte[] payload, QoS qoS, boolean retain) throws Exception {
- connection.publish(topic, payload, qoS, retain, null);
+ void publish(String topic, byte[] payload, QoS qoS, boolean retain, Callback<Void> callback) throws Exception {
+ connection.publish(topic, payload, qoS, retain, callback);
}
void addConsumer(MQTTConsumer consumer) {
@@ -174,6 +167,6 @@ public class MQTTEndpoint extends Defaul
}
public boolean isSingleton() {
- return false;
+ return true;
}
}
Modified: camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTProducer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTProducer.java?rev=1395089&r1=1395088&r2=1395089&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTProducer.java (original)
+++ camel/branches/camel-2.10.x/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTProducer.java Sat Oct 6 15:10:28 2012
@@ -20,6 +20,7 @@ import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultAsyncProducer;
+import org.fusesource.mqtt.client.Callback;
import org.fusesource.mqtt.client.QoS;
public class MQTTProducer extends DefaultAsyncProducer implements Processor {
@@ -29,41 +30,58 @@ public class MQTTProducer extends Defaul
public MQTTProducer(MQTTEndpoint mqttEndpoint) {
super(mqttEndpoint);
this.mqttEndpoint = mqttEndpoint;
-
}
@Override
- public boolean process(Exchange exchange, AsyncCallback asyncCallback) {
- try {
- doProcess(exchange);
- } catch (Exception e) {
- exchange.setException(e);
- }
- asyncCallback.done(true);
- return true;
- }
-
- void doProcess(Exchange exchange) throws Exception {
+ public boolean process(final Exchange exchange, final AsyncCallback callback) {
byte[] body = exchange.getIn().getBody(byte[].class);
if (body != null) {
MQTTConfiguration configuration = mqttEndpoint.getConfiguration();
- boolean retain = configuration.isByDefaultRetain();
+ boolean retain = exchange.getProperty(configuration.getMqttRetainPropertyName(), configuration.isByDefaultRetain(), Boolean.class);
- if (exchange.getProperty(configuration.getMqttRetainPropertyName()) != null) {
- retain = exchange.getProperty(configuration.getMqttRetainPropertyName(), Boolean.class);
- }
QoS qoS = configuration.getQoS();
Object qoSValue = exchange.getProperty(configuration.getMqttQosPropertyName());
if (qoSValue != null) {
qoS = MQTTConfiguration.getQoS(qoSValue.toString());
}
+ // where should we publish to
String topicName = configuration.getPublishTopicName();
Object topicValue = exchange.getProperty(configuration.getMqttTopicPropertyName());
if (topicValue != null) {
topicName = topicValue.toString();
}
- mqttEndpoint.publish(topicName, body, qoS, retain);
+ final String name = topicName;
+
+ try {
+ log.debug("Publishing to {}", name);
+ mqttEndpoint.publish(name, body, qoS, retain, new Callback<Void>() {
+ @Override
+ public void onSuccess(Void aVoid) {
+ log.trace("onSuccess from {}", name);
+ callback.done(false);
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ log.trace("onFailure from {}", name);
+ exchange.setException(throwable);
+ callback.done(false);
+ }
+ });
+ } catch (Exception e) {
+ exchange.setException(e);
+ callback.done(true);
+ return true;
+ }
+
+ // we continue async, as the mqtt endpoint will invoke the callback when its done
+ return false;
+ } else {
+ // no data to send so we are done
+ log.trace("No data to publish");
+ callback.done(true);
+ return true;
}
}
}
Modified: camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTConfigurationTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTConfigurationTest.java?rev=1395089&r1=1395088&r2=1395089&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTConfigurationTest.java (original)
+++ camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTConfigurationTest.java Sat Oct 6 15:10:28 2012
@@ -23,7 +23,6 @@ import org.junit.Test;
public class MQTTConfigurationTest extends MQTTBaseTest {
-
@Test
public void testBasicConfiguration() throws Exception {
Endpoint endpoint = context.getEndpoint("mqtt:todo?byDefaultRetain=true&qualityOfService=exactlyOnce&publishTopicName=" + TEST_TOPIC + "&subscribeTopicName=" + TEST_TOPIC);
Modified: camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTConsumerTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTConsumerTest.java?rev=1395089&r1=1395088&r2=1395089&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTConsumerTest.java (original)
+++ camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTConsumerTest.java Sat Oct 6 15:10:28 2012
@@ -28,10 +28,8 @@ import org.junit.Test;
public class MQTTConsumerTest extends MQTTBaseTest {
-
@Test
public void testConsume() throws Exception {
-
MQTT mqtt = new MQTT();
BlockingConnection publisherConnection = mqtt.blockingConnection();
Topic topic = new Topic(TEST_TOPIC, QoS.AT_MOST_ONCE);
@@ -48,12 +46,13 @@ public class MQTTConsumerTest extends MQ
mock.assertIsSatisfied();
}
-
protected RouteBuilder createRouteBuilder() {
return new RouteBuilder() {
public void configure() {
- from("mqtt:bar?subscribeTopicName=" + TEST_TOPIC).transform(body().convertToString()).to("mock:result");
+ from("mqtt:bar?subscribeTopicName=" + TEST_TOPIC)
+ .transform(body().convertToString())
+ .to("mock:result");
}
};
}
Modified: camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTProducerTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTProducerTest.java?rev=1395089&r1=1395088&r2=1395089&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTProducerTest.java (original)
+++ camel/branches/camel-2.10.x/components/camel-mqtt/src/test/java/org/apache/camel/component/mqtt/MQTTProducerTest.java Sat Oct 6 15:10:28 2012
@@ -33,7 +33,6 @@ public class MQTTProducerTest extends MQ
@Test
public void testProduce() throws Exception {
-
MQTT mqtt = new MQTT();
final BlockingConnection subscribeConnection = mqtt.blockingConnection();
subscribeConnection.connect();
@@ -53,7 +52,6 @@ public class MQTTProducerTest extends MQ
e.printStackTrace();
break;
}
-
}
}
});
@@ -69,9 +67,7 @@ public class MQTTProducerTest extends MQ
assertTrue("Messages not consumed = " + latch.getCount(), latch.getCount() == 0);
}
-
protected RouteBuilder createRouteBuilder() {
-
return new RouteBuilder() {
public void configure() {
from("direct:foo").to("mqtt:boo?publishTopicName=" + TEST_TOPIC);
Modified: camel/branches/camel-2.10.x/components/camel-mqtt/src/test/resources/log4j.properties
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-mqtt/src/test/resources/log4j.properties?rev=1395089&r1=1395088&r2=1395089&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-mqtt/src/test/resources/log4j.properties (original)
+++ camel/branches/camel-2.10.x/components/camel-mqtt/src/test/resources/log4j.properties Sat Oct 6 15:10:28 2012
@@ -21,7 +21,7 @@
log4j.rootLogger=info, file
#log4j.logger.twitter4j=DEBUG
-#log4j.logger.org.apache.camel.component.mqttBUG
+#log4j.logger.org.apache.camel.component.mqtt=DEBUG
#log4j.logger.org.apache.camel=DEBUG
# CONSOLE appender not used by default
@@ -33,5 +33,5 @@ log4j.appender.out.layout.ConversionPatt
log4j.appender.file=org.apache.log4j.FileAppender
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
-log4j.appender.file.file=target/camel-twitter-test.log
+log4j.appender.file.file=target/camel-mqtt-test.log
log4j.appender.file.append=true