Cleanup subscribe listeners after failed subscribe #432

Author: Ian Craggs

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java

@@ -44,6 +44,7 @@
 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
 import org.eclipse.paho.client.mqttv3.util.Debug;
+import org.eclipse.paho.client.mqttv3.IMqttToken;
 
 /**
  * Lightweight client for talking to an MQTT server using non-blocking methods
@@ -1014,7 +1015,7 @@ public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext
 		if ((messageListeners.length != qos.length) || (qos.length != topicFilters.length)) {
 			throw new IllegalArgumentException();
 		}
-		
+
 		// add or remove message handlers to the list for this client
 		for (int i = 0; i < topicFilters.length; ++i) {
 			MqttTopic.validate(topicFilters[i], true/* allow wildcards */);
@@ -1025,8 +1026,18 @@ public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext
                 this.comms.setMessageListener(topicFilters[i], messageListeners[i]);
             }
 		}
-		
-		return this.subscribeBase(topicFilters, qos, userContext, callback);
+
+		IMqttToken token = null;
+		try 	{
+			token = this.subscribeBase(topicFilters, qos, userContext, callback);
+		} catch(Exception e) {
+			// if the subscribe fails, then we have to remove the message handlers
+			for (int i = 0; i < topicFilters.length; ++i) {
+				this.comms.removeMessageListener(topicFilters[i]);
+			}
+			throw e;
+		}
+		return token;
 	}
 
 	/*

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttAsyncClient.java

@@ -1217,7 +1217,17 @@ public IMqttToken subscribe(MqttSubscription[] subscriptions, Object userContext
 			this.comms.setMessageListener(null, subscriptions[i].getTopic(), messageListeners[i]);
 		}
 
-		return this.subscribeBase(subscriptions, userContext, callback, subscriptionProperties);
+		IMqttToken token = null;
+		try 	{
+			token = this.subscribeBase(subscriptions, userContext, callback, subscriptionProperties);
+		} catch(Exception e) {
+			// if the subscribe fails, then we have to remove the message handlers
+			for (int i = 0; i < subscriptions.length; ++i) {
+				this.comms.removeMessageListener(subscriptions[i].getTopic());
+			}
+			throw e;
+		}
+		return token;
 	}
 
 	/*
@@ -1262,7 +1272,17 @@ public IMqttToken subscribe(MqttSubscription[] subscriptions, Object userContext
 			this.comms.setMessageListener(subId, subscriptions[i].getTopic(), messageListener);
 		}
 
-		return this.subscribeBase(subscriptions, userContext, callback, subscriptionProperties);
+		IMqttToken token = null;
+		try 	{
+			token = this.subscribeBase(subscriptions, userContext, callback, subscriptionProperties);
+		} catch(Exception e) {
+			// if the subscribe fails, then we have to remove the message handlers
+			for (int i = 0; i < subscriptions.length; ++i) {
+				this.comms.removeMessageListener(subscriptions[i].getTopic());
+			}
+			throw e;
+		}
+		return token;
 	}
 
 	/*