From 6550cda78962d9ca9ad89f95ce6ea9acb8239a5c Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Mon, 6 Mar 2017 14:34:10 -0500
Subject: [PATCH] AMQP-4238: Detect Subscription Failures and QOS

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

Revert to using the sync client in the message-driven adapter so we can detect
subscription failures (the sync client throws an exception).

The only reason to use the async client was to timeout disconnects; this can
be achieved with the sync client and `disconnectForcibly`.

Also, the subscribe method updates the qos argument with the granted QOS values.

Detect and log if any QOS does not match the request.

Polishing

Polishing - PR Comments

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

* Remove all new tests since Paho lib has class signature check, so we can't mock its classes

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

* Fix only the `subscribe()` bug; leave the `async` client
---
 .../MqttPahoMessageDrivenChannelAdapter.java  | 21 ++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
index d03693decb..6dd6963f7f 100644
--- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
+++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
@@ -13,6 +13,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.integration.mqtt.inbound;
 
 import java.util.Arrays;
@@ -20,6 +21,7 @@ import java.util.concurrent.ScheduledFuture;
 
 import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.IMqttToken;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
@@ -155,7 +157,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 			try {
 				if (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_ALWAYS)
 						|| (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_CLEAN)
-								&& this.cleanSession)) {
+						&& this.cleanSession)) {
 					this.client.unsubscribe(getTopic())
 							.waitForCompletion(this.completionTimeout);
 				}
@@ -187,8 +189,12 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 		try {
 			super.addTopic(topic, qos);
 			if (this.client != null && this.client.isConnected()) {
-				this.client.subscribe(topic, qos)
-						.waitForCompletion(this.completionTimeout);
+				IMqttToken mqttToken = this.client.subscribe(topic, qos);
+				mqttToken.waitForCompletion(this.completionTimeout);
+				int[] grantedQos = mqttToken.getGrantedQos();
+				if (grantedQos != null && grantedQos.length == 1 && grantedQos[0] == 0x80) {
+					throw new MqttException(MqttException.REASON_CODE_SUBSCRIBE_FAILED);
+				}
 			}
 		}
 		catch (MqttException e) {
@@ -211,7 +217,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 			super.removeTopic(topic);
 		}
 		catch (MqttException e) {
-			throw new MessagingException("Failed to unsubscribe from topic " + Arrays.asList(topic), e);
+			throw new MessagingException("Failed to unsubscribe from topic " + Arrays.toString(topic), e);
 		}
 		finally {
 			this.topicLock.unlock();
@@ -230,17 +236,18 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 		this.client.setCallback(this);
 
 		this.topicLock.lock();
+		String[] topics = getTopic();
 		try {
 			this.client.connect(connectionOptions)
 					.waitForCompletion(this.completionTimeout);
-			this.client.subscribe(getTopic(), getQos())
+			this.client.subscribe(topics, getQos())
 					.waitForCompletion(this.completionTimeout);
 		}
 		catch (MqttException e) {
 			if (this.applicationEventPublisher != null) {
 				this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, e));
 			}
-			logger.error("Error connecting or subscribing to " + Arrays.asList(getTopic()), e);
+			logger.error("Error connecting or subscribing to " + Arrays.toString(topics), e);
 			this.client.disconnect()
 					.waitForCompletion(this.completionTimeout);
 			throw e;
@@ -250,7 +257,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 		}
 		if (this.client.isConnected()) {
 			this.connected = true;
-			String message = "Connected and subscribed to " + Arrays.asList(getTopic());
+			String message = "Connected and subscribed to " + Arrays.toString(topics);
 			if (logger.isDebugEnabled()) {
 				logger.debug(message);
 			}