INT-3468 MQTT Async Client

JIRA: https://jira.spring.io/browse/INT-3468

Provide an option to not block when sending and emit
events for sends and delivery confirmations.

Also use the async client for the inbound adapter; while
it doesn't make any performance difference, it does
allow us to timeout the disconnect, which we have seen
to cause hangs on the CI servers.

INT-3468 Polishing - Use Events; Add Docs

INT-3468 Doc Polishing

INT-3468 More Polishing - PR Comments

- Only emit delivered event if async
- Add clientId and a new instance counter to events

INT-3468 Polishing

- Pull client instance up to the abstract class and remove references to the Paho implementation from the events
- Improve tests to include a second client

INT-3468 Fix Test Case

Incorrect classname meant the default location for the application
context config was not found on case-sensitive file systems.

Also, don't auto-start the adapters in the context, in case
the broker is not running.

INT-3468: Polishing

INT-3468 Fix Package Tangle; Add 'async-events'

Add an option (default false) to emit events when async is true.

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java

@@ -44,6 +44,7 @@ public class MqttMessageDrivenChannelAdapterParser extends AbstractChannelAdapte
 		builder.addConstructorArgValue(element.getAttribute("topics"));
 		builder.addPropertyReference("outputChannel", channelName);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "qos");
 
 		return builder.getBeanDefinition();
 	}

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java

@@ -59,6 +59,8 @@ public class MqttOutboundChannelAdapterParser extends AbstractOutboundChannelAda
 		}
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-qos");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-retained");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async-events");
 
 		return builder.getBeanDefinition();
 

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/DefaultMqttPahoClientFactory.java

@@ -19,6 +19,7 @@ import java.util.Properties;
 
 import javax.net.SocketFactory;
 
+import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
@@ -112,6 +113,12 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory {
 		return new MqttClient(uri == null ? "tcp://NO_URL_PROVIDED" : uri, clientId, this.persistence);
 	}
 
+	@Override
+	public MqttAsyncClient getAsyncClientInstance(String uri, String clientId) throws MqttException {
+		// Client validates URI even if overridden by options
+		return new MqttAsyncClient(uri == null ? "tcp://NO_URL_PROVIDED" : uri, clientId, this.persistence);
+	}
+
 	@Override
 	public MqttConnectOptions getConnectionOptions() {
 		MqttConnectOptions options = new MqttConnectOptions();

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttPahoClientFactory.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.integration.mqtt.core;
 
+import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
@@ -36,6 +37,17 @@ public interface MqttPahoClientFactory {
 	 */
 	MqttClient getClientInstance(String url, String clientId) throws MqttException;
 
+	/**
+	 * Retrieve an async client instance.
+	 *
+	 * @param url The URL.
+	 * @param clientId The client id.
+	 * @return The client instance.
+	 * @throws MqttException Any.
+	 * @since 4.1
+	 */
+	MqttAsyncClient getAsyncClientInstance(String url, String clientId) throws MqttException;
+
 	/**
 	 * Retrieve the connection options.
 	 *

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttIntegrationEvent.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed 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.springframework.integration.mqtt.event;
+
+import org.springframework.integration.event.IntegrationEvent;
+
+/**
+ * Base class for Mqtt Events.
+ * @author Gary Russell
+ *
+ * @since 4.1
+ */
+@SuppressWarnings("serial")
+public abstract class MqttIntegrationEvent extends IntegrationEvent {
+
+	public MqttIntegrationEvent(Object source) {
+		super(source);
+	}
+
+	public MqttIntegrationEvent(Object source, Throwable cause) {
+		super(source, cause);
+	}
+
+}

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageDeliveredEvent.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed 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.springframework.integration.mqtt.event;
+
+
+
+/**
+ * An event emitted (when using aysnc) when the client indicates the message
+ * was delivered.
+ *
+ * @author Gary Russell
+ * @since 4.1
+ *
+ */
+@SuppressWarnings("serial")
+public class MqttMessageDeliveredEvent extends MqttMessageDeliveryEvent {
+
+	public MqttMessageDeliveredEvent(Object source, int messageId, String clientId,
+			int clientInstance) {
+		super(source, messageId, clientId, clientInstance);
+	}
+
+	@Override
+	public String toString() {
+		return "MqttMessageSentEvent [clientId=" + getClientId()
+				+ ", clientInstance=" + getClientInstance()
+				+ ", messageId=" + getMessageId()
+				+ "]";
+	}
+
+}

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageDeliveryEvent.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed 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.springframework.integration.mqtt.event;
+
+
+/**
+ * Base class for events related to message delivery. Properties {@link #messageId},
+ * {@link #clientId} and {@link #clientInstance} can be used to correlate events.
+ *
+ * @author Gary Russell
+ * @since 4.1
+ *
+ */
+@SuppressWarnings("serial")
+public abstract class MqttMessageDeliveryEvent extends MqttIntegrationEvent {
+
+	private final int messageId;
+
+	private final String clientId;
+
+	private final int clientInstance;
+
+	public MqttMessageDeliveryEvent(Object source, int messageId, String clientId, int clientInstance) {
+		super(source);
+		this.messageId = messageId;
+		this.clientId = clientId;
+		this.clientInstance = clientInstance;
+	}
+
+	public int getMessageId() {
+		return messageId;
+	}
+
+	public String getClientId() {
+		return clientId;
+	}
+
+	public int getClientInstance() {
+		return clientInstance;
+	}
+
+}

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/MqttMessageSentEvent.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2014 the original author or authors.
+ *
+ * Licensed 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.springframework.integration.mqtt.event;
+
+import org.springframework.messaging.Message;
+
+/**
+ * An event emitted (when using aysnc) when the client indicates that a message
+ * has been sent.
+ *
+ * @author Gary Russell
+ * @since 4.1
+ *
+ */
+@SuppressWarnings("serial")
+public class MqttMessageSentEvent extends MqttMessageDeliveryEvent {
+
+	private final Message<?> message;
+
+	private final String topic;
+
+	public MqttMessageSentEvent(Object source, Message<?> message, String topic, int messageId,
+			String clientId, int clientInstance) {
+		super(source, messageId, clientId, clientInstance);
+		this.message = message;
+		this.topic = topic;
+	}
+
+	public Message<?> getMessage() {
+		return message;
+	}
+
+	public String getTopic() {
+		return topic;
+	}
+
+	@Override
+	public String toString() {
+		return "MqttMessageSentEvent [message=" + message
+				+ ", topic=" + topic
+				+ ", clientId=" + getClientId()
+				+ ", clientInstance=" + getClientInstance()
+				+ ", messageId=" + getMessageId()
+				+ "]";
+	}
+
+}

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/event/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * ApplicationEvents generated by the mqtt module.
+ */
+package org.springframework.integration.mqtt.event;

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java

@@ -35,6 +35,8 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro
 
 	private final String[] topic;
 
+	private volatile int[] qos;
+
 	private volatile MqttMessageConverter converter;
 
 	public AbstractMqttMessageDrivenChannelAdapter(String url, String clientId, String... topic) {
@@ -45,6 +47,16 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro
 		this.url = url;
 		this.clientId = clientId;
 		this.topic = topic;
+		// set the topic qos to 1 by default
+		this.qos = buildQosArray(1);
+	}
+
+	private int[] buildQosArray(int value) {
+		int[] qos = new int[this.topic.length];
+		for (int i = 0; i < qos.length; i++) {
+			qos[i] = value;
+		}
+		return qos;
 	}
 
 	public void setConverter(MqttMessageConverter converter) {
@@ -52,6 +64,27 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro
 		this.converter = converter;
 	}
 
+	/**
+	 * Set the QoS for each topic; a single value will apply to all topics otherwise
+	 * the correct number of qos values must be provided.
+	 * @param qos The qos value(s).
+	 * @since 4.1
+	 */
+	public void setQos(int... qos) {
+		Assert.notNull(qos, "'qos' cannot be null");
+		if (qos.length == 1) {
+			this.qos = buildQosArray(qos[0]);
+		}
+		else {
+			Assert.isTrue(qos.length == this.topic.length);
+			this.qos = qos;
+		}
+	}
+
+	protected int[] getQos() {
+		return qos;
+	}
+
 	protected String getUrl() {
 		return url;
 	}

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

@@ -19,8 +19,8 @@ import java.util.Arrays;
 import java.util.concurrent.ScheduledFuture;
 
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
-import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
@@ -40,14 +40,18 @@ import org.springframework.util.Assert;
 public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDrivenChannelAdapter
 		implements MqttCallback {
 
+	private static final int DEFAULT_COMPLETION_TIMEOUT = 30000;
+
 	private final MqttPahoClientFactory clientFactory;
 
-	private volatile MqttClient client;
+	private volatile MqttAsyncClient client;
 
 	private volatile ScheduledFuture<?> reconnectFuture;
 
 	private volatile boolean connected;
 
+	private volatile int completionTimeout = DEFAULT_COMPLETION_TIMEOUT;
+
 
 	/**
 	 * Use this constructor for a single url (although it may be overridden
@@ -88,6 +92,16 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 		this(url, clientId, new DefaultMqttPahoClientFactory(), topic);
 	}
 
+	/**
+	 * Set the completion timeout for async operations. Not settable using the namespace.
+	 * Default 30000 milliseconds.
+	 * @param completionTimeout The timeout.
+	 * @since 4.1
+	 */
+	public void setCompletionTimeout(int completionTimeout) {
+		this.completionTimeout = completionTimeout;
+	}
+
 	@Override
 	protected void doStart() {
 		super.doStart();
@@ -105,13 +119,15 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 		this.cancelReconnect();
 		super.doStop();
 		try {
-			this.client.unsubscribe(this.getTopic());
+			this.client.unsubscribe(this.getTopic())
+					.waitForCompletion(this.completionTimeout);
 		}
 		catch (MqttException e) {
 			logger.error("Exception while unsubscribing", e);
 		}
 		try {
-			this.client.disconnect();
+			this.client.disconnect()
+					.waitForCompletion(this.completionTimeout);
 		}
 		catch (MqttException e) {
 			logger.error("Exception while disconnecting", e);
@@ -127,18 +143,21 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 	}
 
 	private void connectAndSubscribe() throws MqttException {
-		this.client.setCallback(this);
 		MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions();
 		Assert.state(this.getUrl() != null || connectionOptions.getServerURIs() != null,
 				"If no 'url' provided, connectionOptions.getServerURIs() must not be null");
-		this.client = this.clientFactory.getClientInstance(this.getUrl(), this.getClientId());
-		this.client.connect(connectionOptions);
-
+		this.client = this.clientFactory.getAsyncClientInstance(this.getUrl(), this.getClientId());
+		this.client.setCallback(this);
+		this.client.connect(connectionOptions)
+				.waitForCompletion(this.completionTimeout);
 		try {
-			this.client.subscribe(this.getTopic());
+			this.client.subscribe(this.getTopic(), this.getQos())
+					.waitForCompletion(this.completionTimeout);
 		}
 		catch (MqttException e) {
-			this.client.disconnect();
+			logger.error("Error subscribing to " + Arrays.asList(this.getTopic()), e);
+			this.client.disconnect()
+					.waitForCompletion(this.completionTimeout);
 			throw e;
 		}
 		if (this.client.isConnected()) {

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java

@@ -16,15 +16,13 @@
 
 package org.springframework.integration.mqtt.outbound;
 
-import org.eclipse.paho.client.mqttv3.MqttMessage;
-
 import org.springframework.context.SmartLifecycle;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
 import org.springframework.integration.mqtt.support.MqttHeaders;
-import org.springframework.integration.mqtt.support.MqttMessageConverter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.util.Assert;
 
 /**
@@ -46,7 +44,7 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
 
 	private volatile boolean defaultRetained = false;
 
-	private volatile MqttMessageConverter converter;
+	private volatile MessageConverter converter;
 
 	private boolean running;
 
@@ -54,6 +52,8 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
 
 	private volatile boolean autoStartup;
 
+	private volatile int clientInstance;
+
 	public AbstractMqttMessageHandler(String url, String clientId) {
 		Assert.hasText(clientId, "'clientId' cannot be null or empty");
 		this.url = url;
@@ -72,24 +72,41 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
 		this.defaultRetained = defaultRetain;
 	}
 
-	public void setConverter(MqttMessageConverter converter) {
+	public void setConverter(MessageConverter converter) {
 		Assert.notNull(converter, "'converter' cannot be null");
 		this.converter = converter;
 	}
 
+	protected MessageConverter getConverter() {
+		return converter;
+	}
+
 	protected String getUrl() {
 		return url;
 	}
 
-	protected String getClientId() {
+	public String getClientId() {
 		return clientId;
 	}
 
+	/**
+	 * Incremented each time the client is connected.
+	 * @return The instance;
+	 * @since 4.1
+	 */
+	public int getClientInstance() {
+		return clientInstance;
+	}
+
 	@Override
 	public String getComponentType() {
 		return "mqtt:outbound-channel-adapter";
 	}
 
+	protected void incrementClientInstance() {
+		this.clientInstance++;
+	}
+
 	@Override
 	protected void onInit() throws Exception {
 		super.onInit();
@@ -145,16 +162,16 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
 	protected void handleMessageInternal(Message<?> message) throws Exception {
 		this.connectIfNeeded();
 		String topic = (String) message.getHeaders().get(MqttHeaders.TOPIC);
-		MqttMessage mqttMessage = (MqttMessage) this.converter.fromMessage(message, MqttMessage.class);
+		Object mqttMessage = this.converter.fromMessage(message, Object.class);
 		if (topic == null && this.defaultTopic == null) {
 			throw new MessageHandlingException(message,
 					"No '" + MqttHeaders.TOPIC + "' header and no default topic defined");
 		}
-		this.publish(topic == null ? this.defaultTopic : topic, mqttMessage);
+		this.publish(topic == null ? this.defaultTopic : topic, mqttMessage, message);
 	}
 
 	protected abstract void connectIfNeeded();
 
-	protected abstract void publish(String topic, Object mqttMessage) throws Exception;
+	protected abstract void publish(String topic, Object mqttMessage, Message<?> message) throws Exception;
 
 }

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java

@@ -16,14 +16,20 @@
 package org.springframework.integration.mqtt.outbound;
 
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
-import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
 import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
+import org.springframework.integration.mqtt.event.MqttMessageDeliveredEvent;
+import org.springframework.integration.mqtt.event.MqttMessageSentEvent;
+import org.springframework.integration.mqtt.support.MqttMessageConverter;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
 
@@ -35,11 +41,21 @@ import org.springframework.util.Assert;
  *
  */
 public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
-		implements MqttCallback {
+		implements MqttCallback, ApplicationEventPublisherAware {
+
+	private static final int DEFAULT_COMPLETION_TIMEOUT = 30000;
+
+	private volatile int completionTimeout = DEFAULT_COMPLETION_TIMEOUT;
 
 	private final MqttPahoClientFactory clientFactory;
 
-	private volatile MqttClient client;
+	private volatile MqttAsyncClient client;
+
+	private volatile boolean async;
+
+	private volatile boolean asyncEvents;
+
+	private volatile ApplicationEventPublisher applicationEventPublisher;
 
 	/**
 	 * Use this constructor for a single url (although it may be overridden
@@ -75,6 +91,51 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
 		this(url, clientId, new DefaultMqttPahoClientFactory());
 	}
 
+	/**
+	 * Set to true if you don't want to block when sending messages. Default false.
+	 * When true, message sent/delivered events will be published for reception
+	 * by a suitably configured 'ApplicationListener' or an event
+	 * inbound-channel-adapter.
+	 * @param async true for async.
+	 * @since 4.1
+	 */
+	public void setAsync(boolean async) {
+		this.async = async;
+	}
+
+	/**
+	 * When {@link #setAsync(boolean)} is true, setting this to true enables
+	 * publication of {@link MqttMessageSentEvent} and {@link MqttMessageDeliveredEvent}
+	 * to be emitted. Default false.
+	 * @param asyncEvents the asyncEvents.
+	 * @since 4.1
+	 */
+	public void setAsyncEvents(boolean asyncEvents) {
+		this.asyncEvents = asyncEvents;
+	}
+
+	/**
+	 * Set the completion timeout for async operations. Not settable using the namespace.
+	 * Default 30000 milliseconds.
+	 * @param completionTimeout The timeout.
+	 * @since 4.1
+	 */
+	public void setCompletionTimeout(int completionTimeout) {
+		this.completionTimeout = completionTimeout;
+	}
+
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
+	@Override
+	protected void onInit() throws Exception {
+		super.onInit();
+		Assert.state(getConverter() instanceof MqttMessageConverter,
+				"MessageConverter must be an MqttMessageConverter");
+	}
+
 	@Override
 	protected void doStart() {
 	}
@@ -83,7 +144,7 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
 	protected void doStop() {
 		try {
 			if (this.client != null) {
-				this.client.disconnect();
+				this.client.disconnect().waitForCompletion(this.completionTimeout);
 				this.client.close();
 				this.client = null;
 			}
@@ -102,9 +163,10 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
 			MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions();
 			Assert.state(this.getUrl() != null || connectionOptions.getServerURIs() != null,
 					"If no 'url' provided, connectionOptions.getServerURIs() must not be null");
-			this.client = this.clientFactory.getClientInstance(this.getUrl(), this.getClientId());
-			this.client.connect(connectionOptions);
+			this.client = this.clientFactory.getAsyncClientInstance(this.getUrl(), this.getClientId());
+			incrementClientInstance();
 			this.client.setCallback(this);
+			this.client.connect(connectionOptions).waitForCompletion(this.completionTimeout);
 			if (logger.isDebugEnabled()) {
 				logger.debug("Client connected");
 			}
@@ -124,9 +186,25 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
 	}
 
 	@Override
-	protected void publish(String topic, Object mqttMessage) throws Exception {
+	protected void publish(String topic, Object mqttMessage, Message<?> message) throws Exception {
 		Assert.isInstanceOf(MqttMessage.class, mqttMessage);
-		this.client.publish(topic, (MqttMessage) mqttMessage);
+		IMqttDeliveryToken token = this.client.publish(topic, (MqttMessage) mqttMessage);
+		if (!this.async) {
+			token.waitForCompletion(this.completionTimeout);
+		}
+		else if (this.asyncEvents && this.applicationEventPublisher != null) {
+			this.applicationEventPublisher.publishEvent(
+					new MqttMessageSentEvent(this, message, topic, token.getMessageId(), getClientId(),
+							getClientInstance()));
+		}
+	}
+
+	private void sendDeliveryComplete(IMqttDeliveryToken token) {
+		if (this.async && this.asyncEvents && this.applicationEventPublisher != null) {
+			this.applicationEventPublisher.publishEvent(
+					new MqttMessageDeliveredEvent(this, token.getMessageId(), getClientId(),
+							getClientInstance()));
+		}
 	}
 
 	@Override
@@ -142,7 +220,7 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
 
 	@Override
 	public void deliveryComplete(IMqttDeliveryToken token) {
-
+		sendDeliveryComplete(token);
 	}
 
 }

spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-4.1.xsd

@@ -43,6 +43,15 @@
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="qos">
+				<xsd:annotation>
+					<xsd:documentation>
+						Specifies the QoS to use when subscribing to topics; default '1'. This can be single
+						value (applying to all topics); otherwise it must be a comma-delimited list corresponding
+						to the provided topics (the name number of elements must be provided).
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 			<xsd:attribute name="send-timeout" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
@@ -131,6 +140,27 @@
 						</xsd:documentation>
 					</xsd:annotation>
 				</xsd:attribute>
+				<xsd:attribute name="async">
+					<xsd:annotation>
+						<xsd:documentation>
+							Specifies that sends should not block, with the thread returning
+							immediately the message is sent. When 'true', message
+							sent and message delivery events can be published; see 'async-events'.
+							Default: 'false'.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
+				<xsd:attribute name="async-events">
+					<xsd:annotation>
+						<xsd:documentation>
+							When 'async' is true, specifies that message
+							sent and message delivery events will be published for reception
+							by a suitably configured 'ApplicationListener' or an event
+							inbound-channel-adapter.
+							Default: 'false'.
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
 			</xsd:complexType>
 	</xsd:element>
 

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests-context.xml

@@ -7,7 +7,7 @@
 		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
 		http://www.springframework.org/schema/integration/mqtt http://www.springframework.org/schema/integration/mqtt/spring-integration-mqtt.xsd">
 
-	<int-mqtt:outbound-channel-adapter id="out" client-id="multiOut" 
+	<int-mqtt:outbound-channel-adapter id="out" client-id="multiOut"
 		client-factory="multiUriClientFactory"
 		default-topic="multiServerTests" />
 

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java

@@ -0,0 +1,293 @@
+/*
+ * Copyright 2002-2014 the original author or authors.
+ *
+ * Licensed 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.springframework.integration.mqtt;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+
+import java.io.File;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
+import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
+import org.springframework.integration.mqtt.event.MqttMessageDeliveredEvent;
+import org.springframework.integration.mqtt.event.MqttMessageSentEvent;
+import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
+import org.springframework.integration.mqtt.support.MqttHeaders;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Gary Russell
+ * @since 4.0
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
+public class BackToBackAdapterTests {
+
+	@ClassRule
+	public static final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883);
+
+	@Autowired
+	private MessageChannel out;
+
+	@Autowired
+	private PollableChannel in;
+
+	@Test
+	public void testSingleTopic() {
+		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
+		adapter.setDefaultTopic("mqtt-foo");
+		adapter.setBeanFactory(mock(BeanFactory.class));
+		adapter.afterPropertiesSet();
+		adapter.start();
+		MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883",
+				"si-test-in", "mqtt-foo");
+		QueueChannel outputChannel = new QueueChannel();
+		inbound.setOutputChannel(outputChannel);
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+		inbound.setTaskScheduler(taskScheduler);
+		inbound.setBeanFactory(mock(BeanFactory.class));
+		inbound.afterPropertiesSet();
+		inbound.start();
+		adapter.handleMessage(new GenericMessage<String>("foo"));
+		adapter.stop();
+		Message<?> out = outputChannel.receive(1000);
+		assertNotNull(out);
+		inbound.stop();
+		assertEquals("foo", out.getPayload());
+		assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC));
+	}
+
+	@Test
+	public void testTwoTopics() {
+		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
+		adapter.setDefaultTopic("mqtt-foo");
+		adapter.setBeanFactory(mock(BeanFactory.class));
+		adapter.afterPropertiesSet();
+		adapter.start();
+		MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883",
+				"si-test-in", "mqtt-foo", "mqtt-bar");
+		QueueChannel outputChannel = new QueueChannel();
+		inbound.setOutputChannel(outputChannel);
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+		inbound.setTaskScheduler(taskScheduler);
+		inbound.setBeanFactory(mock(BeanFactory.class));
+		inbound.afterPropertiesSet();
+		inbound.start();
+		adapter.handleMessage(new GenericMessage<String>("foo"));
+		Message<?> message = MessageBuilder.withPayload("bar").setHeader(MqttHeaders.TOPIC, "mqtt-bar").build();
+		adapter.handleMessage(message);
+		adapter.stop();
+		Message<?> out = outputChannel.receive(1000);
+		assertNotNull(out);
+		inbound.stop();
+		assertEquals("foo", out.getPayload());
+		assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC));
+		out = outputChannel.receive(1000);
+		assertNotNull(out);
+		inbound.stop();
+		assertEquals("bar", out.getPayload());
+		assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC));
+	}
+
+	@Test
+	public void testAsync() throws Exception {
+		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
+		adapter.setDefaultTopic("mqtt-foo");
+		adapter.setBeanFactory(mock(BeanFactory.class));
+		adapter.setAsync(true);
+		adapter.setAsyncEvents(true);
+		EventPublisher publisher = new EventPublisher();
+		adapter.setApplicationEventPublisher(publisher);
+		adapter.afterPropertiesSet();
+		adapter.start();
+		MqttPahoMessageDrivenChannelAdapter inbound =
+				new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in", "mqtt-foo");
+		QueueChannel outputChannel = new QueueChannel();
+		inbound.setOutputChannel(outputChannel);
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+		inbound.setTaskScheduler(taskScheduler);
+		inbound.setBeanFactory(mock(BeanFactory.class));
+		inbound.afterPropertiesSet();
+		inbound.start();
+		GenericMessage<String> message = new GenericMessage<String>("foo");
+		adapter.handleMessage(message);
+		verifyEvents(adapter, publisher, message);
+		adapter.stop();
+		Message<?> out = outputChannel.receive(10000);
+		assertNotNull(out);
+		inbound.stop();
+		assertEquals("foo", out.getPayload());
+		assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC));
+	}
+
+	@Test
+	public void testAsyncPersisted() throws Exception {
+		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
+		String tmpDir = System.getProperty("java.io.tmpdir") + File.separator + "mqtt_persist";
+		new File(tmpDir).mkdirs();
+		MqttClientPersistence persistence = new MqttDefaultFilePersistence(tmpDir);
+		factory.setPersistence(persistence);
+		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out", factory);
+		adapter.setDefaultTopic("mqtt-foo");
+		adapter.setBeanFactory(mock(BeanFactory.class));
+		adapter.setAsync(true);
+		adapter.setAsyncEvents(true);
+		adapter.setDefaultQos(1);
+		EventPublisher publisher1 = new EventPublisher();
+		adapter.setApplicationEventPublisher(publisher1);
+		adapter.afterPropertiesSet();
+		adapter.start();
+
+		MqttPahoMessageDrivenChannelAdapter inbound =
+				new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in", "mqtt-foo", "mqtt-bar");
+		QueueChannel outputChannel = new QueueChannel();
+		inbound.setOutputChannel(outputChannel);
+		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+		inbound.setTaskScheduler(taskScheduler);
+		inbound.setBeanFactory(mock(BeanFactory.class));
+		inbound.afterPropertiesSet();
+		inbound.start();
+		Message<String> message1 = new GenericMessage<String>("foo");
+		adapter.handleMessage(message1);
+		verifyEvents(adapter, publisher1, message1);
+
+		Message<String> message2 = MessageBuilder.withPayload("bar")
+				.setHeader(MqttHeaders.TOPIC, "mqtt-bar")
+				.build();
+		EventPublisher publisher2 = new EventPublisher();
+		adapter.setApplicationEventPublisher(publisher2);
+		adapter.handleMessage(message2);
+		verifyEvents(adapter, publisher2, message2);
+
+		verifyMessageIds(publisher1, publisher2);
+		int clientInstance = publisher1.delivered.getClientInstance();
+
+		adapter.stop();
+		adapter.start(); // new client instance
+
+		publisher1 = new EventPublisher();
+		adapter.setApplicationEventPublisher(publisher1);
+		adapter.handleMessage(message1);
+		verifyEvents(adapter, publisher1, message1);
+
+		publisher2 = new EventPublisher();
+		adapter.setApplicationEventPublisher(publisher2);
+		adapter.handleMessage(message2);
+		verifyEvents(adapter, publisher2, message2);
+
+		verifyMessageIds(publisher1, publisher2);
+
+		assertNotEquals(clientInstance, publisher1.delivered.getClientInstance());
+		adapter.stop();
+
+		Message<?> out = null;
+		for (int i = 0; i < 4; i++) {
+			out = outputChannel.receive(10000);
+			assertNotNull(out);
+			if ("foo".equals(out.getPayload())) {
+				assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC));
+			}
+			else if ("bar".equals(out.getPayload())) {
+				assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC));
+			}
+			else {
+				fail("unexpected payload " + out.getPayload());
+			}
+		}
+		inbound.stop();
+	}
+
+	private void verifyEvents(MqttPahoMessageHandler adapter, EventPublisher publisher1, Message<String> message1)
+			throws InterruptedException {
+		assertTrue(publisher1.latch.await(10, TimeUnit.SECONDS));
+		assertNotNull(publisher1.sent);
+		assertNotNull(publisher1.delivered);
+		assertEquals(publisher1.sent.getMessageId(), publisher1.delivered.getMessageId());
+		assertEquals(adapter.getClientId(), publisher1.sent.getClientId());
+		assertEquals(adapter.getClientId(), publisher1.delivered.getClientId());
+		assertEquals(adapter.getClientInstance(), publisher1.sent.getClientInstance());
+		assertEquals(adapter.getClientInstance(), publisher1.delivered.getClientInstance());
+		assertSame(message1, publisher1.sent.getMessage());
+	}
+
+	private void verifyMessageIds(EventPublisher publisher1, EventPublisher publisher2) {
+		assertNotEquals(publisher1.delivered.getMessageId(), publisher2.delivered.getMessageId());
+		assertEquals(publisher1.delivered.getClientId(), publisher2.delivered.getClientId());
+		assertEquals(publisher1.delivered.getClientInstance(), publisher2.delivered.getClientInstance());
+	}
+
+	@Test
+	public void testMultiURIs() {
+		out.send(new GenericMessage<String>("foo"));
+		Message<?> message = in.receive(10000);
+		assertNotNull(message);
+		assertEquals("foo", message.getPayload());
+	}
+
+	private class EventPublisher implements ApplicationEventPublisher {
+
+		private volatile MqttMessageDeliveredEvent delivered;
+
+		private MqttMessageSentEvent sent;
+
+		private final CountDownLatch latch = new CountDownLatch(2);
+
+		@Override
+		public void publishEvent(ApplicationEvent event) {
+			if (event instanceof MqttMessageSentEvent) {
+				this.sent = (MqttMessageSentEvent) event;
+			}
+			else if (event instanceof MqttMessageDeliveredEvent){
+				this.delivered = (MqttMessageDeliveredEvent) event;
+			}
+			latch.countDown();
+		}
+
+	}
+
+}

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackTobackAdapterTests.java

@@ -1,129 +0,0 @@
-/*
- * Copyright 2002-2014 the original author or authors.
- *
- * Licensed 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.springframework.integration.mqtt;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.mockito.Mockito.mock;
-
-import org.junit.Ignore;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-
-import org.springframework.beans.factory.BeanFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
-import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
-import org.springframework.integration.mqtt.support.MqttHeaders;
-import org.springframework.integration.support.MessageBuilder;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.PollableChannel;
-import org.springframework.messaging.support.GenericMessage;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
-import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
-
-/**
- * @author Gary Russell
- * @since 4.0
- *
- */
-@Ignore //TODO transiently
-@ContextConfiguration
-@RunWith(SpringJUnit4ClassRunner.class)
-@DirtiesContext
-public class BackTobackAdapterTests {
-
-	@Rule
-	public final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883);
-
-	@Autowired
-	public MessageChannel out;
-
-	@Autowired
-	public PollableChannel in;
-
-	@Test
-	public void testSingleTopic() {
-		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
-		adapter.setDefaultTopic("mqtt-foo");
-		adapter.setBeanFactory(mock(BeanFactory.class));
-		adapter.afterPropertiesSet();
-		adapter.start();
-		MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in", "mqtt-foo");
-		QueueChannel outputChannel = new QueueChannel();
-		inbound.setOutputChannel(outputChannel);
-		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
-		taskScheduler.initialize();
-		inbound.setTaskScheduler(taskScheduler);
-		inbound.setBeanFactory(mock(BeanFactory.class));
-		inbound.afterPropertiesSet();
-		inbound.start();
-		adapter.handleMessage(new GenericMessage<String>("foo"));
-		adapter.stop();
-		Message<?> out = outputChannel.receive(1000);
-		assertNotNull(out);
-		inbound.stop();
-		assertEquals("foo", out.getPayload());
-		assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC));
-		adapter.stop();
-	}
-
-	@Test
-	public void testTwoTopics() {
-		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
-		adapter.setDefaultTopic("mqtt-foo");
-		adapter.setBeanFactory(mock(BeanFactory.class));
-		adapter.afterPropertiesSet();
-		adapter.start();
-		MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in", "mqtt-foo", "mqtt-bar");
-		QueueChannel outputChannel = new QueueChannel();
-		inbound.setOutputChannel(outputChannel);
-		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
-		taskScheduler.initialize();
-		inbound.setTaskScheduler(taskScheduler);
-		inbound.setBeanFactory(mock(BeanFactory.class));
-		inbound.afterPropertiesSet();
-		inbound.start();
-		adapter.handleMessage(new GenericMessage<String>("foo"));
-		Message<?> message = MessageBuilder.withPayload("bar").setHeader(MqttHeaders.TOPIC, "mqtt-bar").build();
-		adapter.handleMessage(message);
-		adapter.stop();
-		Message<?> out = outputChannel.receive(1000);
-		assertNotNull(out);
-		inbound.stop();
-		assertEquals("foo", out.getPayload());
-		assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC));
-		out = outputChannel.receive(1000);
-		assertNotNull(out);
-		inbound.stop();
-		assertEquals("bar", out.getPayload());
-		assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC));
-		adapter.stop();
-	}
-
-	@Test
-	public void testMultiURIs() {
-		out.send(new GenericMessage<String>("foo"));
-		Message<?> message = in.receive(10000);
-		assertNotNull(message);
-		assertEquals("foo", message.getPayload());
-	}
-}

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests.java

@@ -32,7 +32,6 @@ import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
 import org.junit.ClassRule;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.invocation.InvocationOnMock;
@@ -58,7 +57,6 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
 @DirtiesContext
-@Ignore //TODO transiently
 public class DownstreamExceptionTests {
 
 	@ClassRule

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java

@@ -22,6 +22,7 @@ import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
@@ -34,10 +35,12 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import javax.net.SocketFactory;
 
+import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
-import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.MqttToken;
 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 import org.junit.Test;
 import org.mockito.invocation.InvocationOnMock;
@@ -107,25 +110,27 @@ public class MqttAdapterTests {
 		factory.setWill(will);
 
 		factory = spy(factory);
-		final MqttClient client = mock(MqttClient.class);
-		doAnswer(new Answer<MqttClient>() {
+		final MqttAsyncClient client = mock(MqttAsyncClient.class);
+		doAnswer(new Answer<MqttAsyncClient>() {
 
 			@Override
-			public MqttClient answer(InvocationOnMock invocation) throws Throwable {
+			public MqttAsyncClient answer(InvocationOnMock invocation) throws Throwable {
 				return client;
 			}
-		}).when(factory).getClientInstance(anyString(), anyString());
+		}).when(factory).getAsyncClientInstance(anyString(), anyString());
 
 		MqttPahoMessageHandler handler = new MqttPahoMessageHandler("foo", "bar", factory);
 		handler.setDefaultTopic("mqtt-foo");
 		handler.setBeanFactory(mock(BeanFactory.class));
 		handler.afterPropertiesSet();
 		handler.start();
+
+		final MqttToken token = mock(MqttToken.class);
 		final AtomicBoolean connectCalled = new AtomicBoolean();
-		doAnswer(new Answer<Object>(){
+		doAnswer(new Answer<MqttToken>(){
 
 			@Override
-			public Object answer(InvocationOnMock invocation) throws Throwable {
+			public MqttToken answer(InvocationOnMock invocation) throws Throwable {
 				MqttConnectOptions options = (MqttConnectOptions) invocation.getArguments()[0];
 				assertEquals(23, options.getConnectionTimeout());
 				assertEquals(45, options.getKeepAliveInterval());
@@ -137,19 +142,22 @@ public class MqttAdapterTests {
 				assertEquals("bar", new String(options.getWillMessage().getPayload()));
 				assertEquals(2, options.getWillMessage().getQos());
 				connectCalled.set(true);
-				return null;
+				return token;
 			}
 		}).when(client).connect(any(MqttConnectOptions.class));
+		doReturn(token).when(client).subscribe(any(String[].class), any(int[].class));
+
+		final MqttDeliveryToken deliveryToken = mock(MqttDeliveryToken.class);
 		final AtomicBoolean publishCalled = new AtomicBoolean();
-		doAnswer(new Answer<Object>() {
+		doAnswer(new Answer<MqttDeliveryToken>() {
 
 			@Override
-			public Object answer(InvocationOnMock invocation) throws Throwable {
+			public MqttDeliveryToken answer(InvocationOnMock invocation) throws Throwable {
 				assertEquals("mqtt-foo", invocation.getArguments()[0]);
 				MqttMessage message = (MqttMessage) invocation.getArguments()[1];
 				assertEquals("Hello, world!", new String(message.getPayload()));
 				publishCalled.set(true);
-				return null;
+				return deliveryToken;
 			}
 		}).when(client).publish(anyString(), any(MqttMessage.class));
 
@@ -177,15 +185,16 @@ public class MqttAdapterTests {
 		factory.setWill(will);
 
 		factory = spy(factory);
-		final MqttClient client = mock(MqttClient.class);
-		doAnswer(new Answer<MqttClient>() {
+		final MqttAsyncClient client = mock(MqttAsyncClient.class);
+		doAnswer(new Answer<MqttAsyncClient>() {
 
 			@Override
-			public MqttClient answer(InvocationOnMock invocation) throws Throwable {
+			public MqttAsyncClient answer(InvocationOnMock invocation) throws Throwable {
 				return client;
 			}
-		}).when(factory).getClientInstance(anyString(), anyString());
+		}).when(factory).getAsyncClientInstance(anyString(), anyString());
 
+		final MqttToken token = mock(MqttToken.class);
 		final AtomicBoolean connectCalled = new AtomicBoolean();
 		doAnswer(new Answer<Object>() {
 
@@ -202,9 +211,10 @@ public class MqttAdapterTests {
 				assertEquals("bar", new String(options.getWillMessage().getPayload()));
 				assertEquals(2, options.getWillMessage().getQos());
 				connectCalled.set(true);
-				return null;
+				return token;
 			}
 		}).when(client).connect(any(MqttConnectOptions.class));
+		doReturn(token).when(client).subscribe(any(String[].class), any(int[].class));
 
 		final AtomicReference<MqttCallback> callback = new AtomicReference<MqttCallback>();
 		doAnswer(new Answer<Object>() {

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests-context.xml

@@ -28,6 +28,19 @@
 		client-id="foo"
 		url="tcp://localhost:1883"
 		topics="bar, baz"
+		qos="0, 2"
+		converter="myConverter"
+		client-factory="clientFactory"
+		send-timeout="123"
+		channel="out" />
+
+	<int-mqtt:message-driven-channel-adapter id="twoTopicsSingleQosAdapter"
+		auto-startup="false"
+		phase="25"
+		client-id="foo"
+		url="tcp://localhost:1883"
+		topics="bar, baz"
+		qos="0"
 		converter="myConverter"
 		client-factory="clientFactory"
 		send-timeout="123"

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests.java

@@ -46,6 +46,9 @@ public class MqttMessageDrivenChannelAdapterParserTests {
 	@Autowired
 	private MqttPahoMessageDrivenChannelAdapter twoTopicsAdapter;
 
+	@Autowired
+	private MqttPahoMessageDrivenChannelAdapter twoTopicsSingleQosAdapter;
+
 	@Autowired
 	private MessageChannel out;
 
@@ -65,6 +68,7 @@ public class MqttMessageDrivenChannelAdapterParserTests {
 		assertEquals(25, TestUtils.getPropertyValue(oneTopicAdapter, "phase"));
 		assertEquals("foo", TestUtils.getPropertyValue(oneTopicAdapter, "clientId"));
 		assertEquals("bar", TestUtils.getPropertyValue(oneTopicAdapter, "topic", String[].class)[0]);
+		assertEquals(1, TestUtils.getPropertyValue(oneTopicAdapter, "qos", int[].class)[0]);
 		assertSame(converter, TestUtils.getPropertyValue(oneTopicAdapter, "converter"));
 		assertEquals(123L, TestUtils.getPropertyValue(oneTopicAdapter, "messagingTemplate.sendTimeout"));
 		assertSame(out, TestUtils.getPropertyValue(oneTopicAdapter, "outputChannel"));
@@ -74,16 +78,26 @@ public class MqttMessageDrivenChannelAdapterParserTests {
 
 	@Test
 	public void testTwoTopics() {
-		assertEquals("tcp://localhost:1883", TestUtils.getPropertyValue(oneTopicAdapter, "url"));
+		assertEquals("tcp://localhost:1883", TestUtils.getPropertyValue(twoTopicsAdapter, "url"));
 		assertFalse(TestUtils.getPropertyValue(twoTopicsAdapter, "autoStartup", Boolean.class));
 		assertEquals(25, TestUtils.getPropertyValue(twoTopicsAdapter, "phase"));
 		assertEquals("foo", TestUtils.getPropertyValue(twoTopicsAdapter, "clientId"));
 		assertEquals("bar", TestUtils.getPropertyValue(twoTopicsAdapter, "topic", String[].class)[0]);
 		assertEquals("baz", TestUtils.getPropertyValue(twoTopicsAdapter, "topic", String[].class)[1]);
+		assertEquals(0, TestUtils.getPropertyValue(twoTopicsAdapter, "qos", int[].class)[0]);
+		assertEquals(2, TestUtils.getPropertyValue(twoTopicsAdapter, "qos", int[].class)[1]);
 		assertSame(converter, TestUtils.getPropertyValue(twoTopicsAdapter, "converter"));
 		assertEquals(123L, TestUtils.getPropertyValue(twoTopicsAdapter, "messagingTemplate.sendTimeout"));
 		assertSame(out, TestUtils.getPropertyValue(twoTopicsAdapter, "outputChannel"));
 		assertSame(clientFactory, TestUtils.getPropertyValue(twoTopicsAdapter, "clientFactory"));
 	}
 
+	@Test
+	public void testTwoTopicsSingleQos() {
+		assertEquals("bar", TestUtils.getPropertyValue(twoTopicsSingleQosAdapter, "topic", String[].class)[0]);
+		assertEquals("baz", TestUtils.getPropertyValue(twoTopicsSingleQosAdapter, "topic", String[].class)[1]);
+		assertEquals(0, TestUtils.getPropertyValue(twoTopicsSingleQosAdapter, "qos", int[].class)[0]);
+		assertEquals(0, TestUtils.getPropertyValue(twoTopicsSingleQosAdapter, "qos", int[].class)[1]);
+	}
+
 }

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests-context.xml

@@ -35,6 +35,8 @@
 			client-factory="clientFactory"
 			phase="25"
 			order="1"
+			async="true"
+			async-events="true"
 			channel="target" />
 
 	<bean id="myConverter" class="org.springframework.integration.mqtt.support.DefaultPahoMessageConverter" />

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java

@@ -16,7 +16,11 @@
 
 package org.springframework.integration.mqtt.config.xml;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 
 import org.hamcrest.Matchers;
 import org.junit.Test;
@@ -70,6 +74,8 @@ public class MqttOutboundChannelAdapterParserTests {
 		assertEquals("bar", TestUtils.getPropertyValue(withConverterHandler, "defaultTopic"));
 		assertSame(converter, TestUtils.getPropertyValue(withConverterHandler, "converter"));
 		assertSame(clientFactory, TestUtils.getPropertyValue(withConverterHandler, "clientFactory"));
+		assertFalse(TestUtils.getPropertyValue(withConverterHandler, "async", Boolean.class));
+		assertFalse(TestUtils.getPropertyValue(withConverterHandler, "asyncEvents", Boolean.class));
 
 		Object handler = TestUtils.getPropertyValue(this.withConverterEndpoint, "handler");
 
@@ -96,6 +102,8 @@ public class MqttOutboundChannelAdapterParserTests {
 		assertEquals(1, TestUtils.getPropertyValue(defaultConverter, "defaultQos"));
 		assertTrue(TestUtils.getPropertyValue(defaultConverter, "defaultRetained", Boolean.class));
 		assertSame(clientFactory, TestUtils.getPropertyValue(withDefaultConverterHandler, "clientFactory"));
+		assertTrue(TestUtils.getPropertyValue(withDefaultConverterHandler, "async", Boolean.class));
+		assertTrue(TestUtils.getPropertyValue(withDefaultConverterHandler, "asyncEvents", Boolean.class));
 	}
 
 }

src/reference/docbook/mqtt.xml

@@ -9,7 +9,7 @@
 			Spring Integration provides inbound and outbound channel adapters supporting the
 			MQ Telemetry Transport (MQTT) protocol. The current implementation uses the
 			<ulink url="http://www.eclipse.org/paho/">Eclipse Paho MQTT Client</ulink>
-			library. 
+			library.
 		</para>
 		<para>
 			Configuration of both adapters is achieved using the
@@ -44,7 +44,8 @@
 		<programlisting><![CDATA[<int-mqtt:message-driven-channel-adapter id="oneTopicAdapter"
 	client-id="foo"]]> <co id="mqtt-i-01"/><![CDATA[
 	url="tcp://localhost:1883"]]> <co id="mqtt-i-02"/><![CDATA[
-	topics="bar"]]> <co id="mqtt-i-03"/><![CDATA[
+	topics="bar,baz"]]> <co id="mqtt-i-03"/><![CDATA[
+	qos="1,2"]]> <co id="mqtt-i-03a"/><![CDATA[
 	converter="myConverter"]]> <co id="mqtt-i-04"/><![CDATA[
 	client-factory="clientFactory"]]> <co id="mqtt-i-05"/><![CDATA[
 	send-timeout="123"]]> <co id="mqtt-i-06"/><![CDATA[
@@ -68,6 +69,10 @@
 			<callout arearefs="mqtt-i-03">
 				A comma delimited list of topics from which this adapter will receive messages.
 			</callout>
+			<callout arearefs="mqtt-i-03a">
+				A comma delimited list of QoS values. Can be a single value that is applied to all
+				topics, or a value for each topic (in which case the lists must the same length).
+			</callout>
 			<callout arearefs="mqtt-i-04">
 				An <interfacename>MqttMessageConverter</interfacename> (optional). The default
 				<classname>DefaultPahoMessageConverter</classname> produces a message with a <code>String</code>
@@ -103,6 +108,11 @@
 			is wrapped in a <code>ConsumerEndpoint</code>. For convenience, it
 			can be configured using the namespace.
 		</para>
+		<para>
+			Starting with <emphasis>version 4.1</emphasis>, the adapter supports asynchronous sends, avoiding
+			blocking until the delivery is confirmed; application events can be emitted to enable applications
+			to confirm delivery if desired.
+		</para>
 		<para>
 			Attributes:
 		</para>
@@ -114,6 +124,8 @@
 	default-qos="1"]]> <co id="mqtt-o-05"/><![CDATA[
 	default-retained="true"]]> <co id="mqtt-o-06"/><![CDATA[
 	default-topic="bar"]]> <co id="mqtt-o-07"/><![CDATA[
+	async="false"]]> <co id="mqtt-o-08"/><![CDATA[
+	async-events="false"]]> <co id="mqtt-o-09"/><![CDATA[
 	channel="target" />]]></programlisting>
 		<calloutlist>
 			<callout arearefs="mqtt-o-01">
@@ -132,7 +144,7 @@
 			</callout>
 			<callout arearefs="mqtt-o-03">
 				An <interfacename>MqttMessageConverter</interfacename> (optional). The default
-				<classname>DefaultPahoMessageConverter</classname> 
+				<classname>DefaultPahoMessageConverter</classname>
 				recognizes the following headers:
 				<itemizedlist>
 					<listitem><code>mqtt_topic</code> - the topic to which the message will be sent</listitem>
@@ -154,6 +166,25 @@
 			<callout arearefs="mqtt-o-07">
 				The default topic to which the message will be sent (used if no <code>mqtt_topic</code> header is found).
 			</callout>
+			<callout arearefs="mqtt-o-08">
+				When <code>true</code>, the caller will not block waiting for delivery confirmation when a message is
+				sent.
+				Default:false (the send blocks until delivery is confirmed).
+			</callout>
+			<callout arearefs="mqtt-o-09">
+				When <code>async</code> and <code>async-events</code> are both <code>true</code>, an
+				<classname>MqttMessageSentEvent</classname> is emitted, containing the message, the
+				topic, the <code>messageId</code> generated by the client library, the <code>clientId</code>
+				and the <code>clientInstance</code> (incremented each time the client is connected).
+				When the delivery is confirmed by the client library, an
+				<classname>MqttMessageDeliveredEvent</classname> is emitted, containing the the <code>messageId</code>,
+				<code>clientId</code> and the <code>clientInstance</code>, enabling
+				delivery to be correlated with the send. These events can be received by any
+				<interfacename>ApplicationListener</interfacename>, or by an event inbound channel adapter. Note that
+				it is possible that the <classname>MqttMessageDeliveredEvent</classname> might be received before
+				the <classname>MqttMessageSentEvent</classname>.
+				Default: <code>false</code>.
+			</callout>
 		</calloutlist>
 	</section>
 

src/reference/docbook/whats-new.xml

@@ -54,13 +54,21 @@
 				See <xref linkend="http-namespace"/> for more information.
 			</para>
 		</section>
-		<section id="4.1-mqtt-cluster">
+		<section id="4.1-mqtt">
 			<title>MQTT Adapter Changes</title>
 			<para>
 				The MQTT channel adapters can now be configured to connect to multiple servers,
 				for example, to support High Availability (HA).
 				See <xref linkend="mqtt"/> for more information.
 			</para>
+			<para>
+				The MQTT message-driven channel adapter now supports specifying the QoS setting for each
+				subscription. See <xref linkend="mqtt"/> for more information.
+			</para>
+			<para>
+				The MQTT outbound channel adapter now supports asynchronous sends, avoiding blocking
+				until delivery is confirmed. See <xref linkend="mqtt"/> for more information.
+			</para>
 		</section>
 	</section>
 </chapter>