From cef58dae340d67d54b647eed7e1ac525dee90714 Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@gopivotal.com>
Date: Wed, 9 Jul 2014 11:21:23 -0400
Subject: [PATCH] INT-3469 MQTT - Set Callback Before Subscribing

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

Previously, the callback was set immediately after
subscribing which caused a small possibility of
message loss.

Set the callback before subscribing.
---
 .../mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java       | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

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 e2889aa378..c4db348019 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
@@ -97,6 +97,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 
 	private void connectAndSubscribe() throws MqttException {
 		this.client = this.clientFactory.getClientInstance(this.getUrl(), this.getClientId());
+		this.client.setCallback(this);
 		this.client.connect(this.clientFactory.getConnectionOptions());
 		try {
 			this.client.subscribe(this.getTopic());
@@ -106,7 +107,6 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 			throw e;
 		}
 		if (this.client.isConnected()) {
-			this.client.setCallback(this);
 			this.connected = true;
 			if (this.reconnectFuture != null) {
 				this.cancelReconnect();