Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQP-4238: Detect Subscription Failures and QOS

Browse Source 
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
This commit is contained in:
Gary Russell
2017-03-06 14:34:10 -05:00
committed by Artem Bilan
parent 1bbc8a6500
commit 6550cda789
1 changed files with 14 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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);
}