You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ni...@apache.org on 2014/11/17 04:22:19 UTC
camel git commit: CAMEL-7662 MQTTProducerTest fails once enables it
Repository: camel
Updated Branches:
refs/heads/master db2b2c33e -> f686fcb0d
CAMEL-7662 MQTTProducerTest fails once enables it
The assertion expects MQTT publish/disconnect call to be submitted as a async task, i.e. enqueued in a dispatch queue
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f686fcb0
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f686fcb0
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f686fcb0
Branch: refs/heads/master
Commit: f686fcb0d3c1a97a561474eb75d934310b8f9e44
Parents: db2b2c3
Author: Tomohisa Igarashi <tm...@gmail.com>
Authored: Sun Nov 16 16:55:45 2014 +0900
Committer: Tomohisa Igarashi <tm...@gmail.com>
Committed: Sun Nov 16 18:57:30 2014 +0900
----------------------------------------------------------------------
components/camel-mqtt/pom.xml | 2 --
.../camel/component/mqtt/MQTTEndpoint.java | 30 +++++++++++++-------
2 files changed, 20 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/f686fcb0/components/camel-mqtt/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-mqtt/pom.xml b/components/camel-mqtt/pom.xml
index 240a0ce..3c7af31 100644
--- a/components/camel-mqtt/pom.xml
+++ b/components/camel-mqtt/pom.xml
@@ -80,8 +80,6 @@
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<forkMode>perTest</forkMode>
- <!--CAMEL-7662 disabling the assertion this time-->
- <enableAssertions>false</enableAssertions>
</configuration>
</plugin>
<plugin>
http://git-wip-us.apache.org/repos/asf/camel/blob/f686fcb0/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java b/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
index 07014ad..664116f 100644
--- a/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
+++ b/components/camel-mqtt/src/main/java/org/apache/camel/component/mqtt/MQTTEndpoint.java
@@ -27,6 +27,7 @@ import org.apache.camel.Producer;
import org.apache.camel.impl.DefaultEndpoint;
import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
+import org.fusesource.hawtdispatch.Task;
import org.fusesource.mqtt.client.Callback;
import org.fusesource.mqtt.client.CallbackConnection;
import org.fusesource.mqtt.client.Listener;
@@ -114,15 +115,19 @@ public class MQTTEndpoint extends DefaultEndpoint {
protected void doStop() throws Exception {
if (connection != null) {
final Promise<Void> promise = new Promise<Void>();
- connection.disconnect(new Callback<Void>() {
- public void onSuccess(Void value) {
- promise.onSuccess(value);
- }
+ connection.getDispatchQueue().execute(new Task() {
+ @Override
+ public void run() {
+ connection.disconnect(new Callback<Void>() {
+ public void onSuccess(Void value) {
+ promise.onSuccess(value);
+ }
- public void onFailure(Throwable value) {
- promise.onFailure(value);
- }
- });
+ public void onFailure(Throwable value) {
+ promise.onFailure(value);
+ }
+ });
+ }});
promise.await(configuration.getDisconnectWaitInSeconds(), TimeUnit.SECONDS);
}
super.doStop();
@@ -169,8 +174,13 @@ public class MQTTEndpoint extends DefaultEndpoint {
return connected;
}
- void publish(String topic, byte[] payload, QoS qoS, boolean retain, Callback<Void> callback) throws Exception {
- connection.publish(topic, payload, qoS, retain, callback);
+ void publish(final String topic, final byte[] payload, final QoS qoS, final boolean retain, final Callback<Void> callback) throws Exception {
+ connection.getDispatchQueue().execute(new Task() {
+ @Override
+ public void run() {
+ connection.publish(topic, payload, qoS, retain, callback);
+ }
+ });
}
void addConsumer(MQTTConsumer consumer) {