Refactor MQTTClient so it relies on MQTTAsync fixes for #432

Author: Ian Craggs

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

@@ -1008,14 +1008,14 @@ public IMqttToken subscribe(String[] topicFilters, int[] qos, IMqttMessageListen
 	public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext, IMqttActionListener callback,
 			IMqttMessageListener[] messageListeners) throws MqttException {
 
-		if ((messageListeners.length != qos.length) || (qos.length != topicFilters.length)) {
+		if (messageListeners != null && (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 */);
-            if (messageListeners[i] == null) {
+            if (messageListeners == null || messageListeners[i] == null) {
                 this.comms.removeMessageListener(topicFilters[i]);
             }
             else {

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

@@ -426,15 +426,7 @@ public void subscribe(String topicFilter, int qos) throws MqttException {
 	 * @see IMqttClient#subscribe(String[], int[])
 	 */
 	public void subscribe(String[] topicFilters, int[] qos) throws MqttException {
-		IMqttToken tok = aClient.subscribe(topicFilters, qos, null, null);
-		tok.waitForCompletion(getTimeToWait());
-		int[] grantedQos = tok.getGrantedQos();
-		for (int i = 0; i < grantedQos.length; ++i) {
-			qos[i] = grantedQos[i];
-		}
-		if (grantedQos.length == 1 && qos[0] == 0x80) {
-			throw new MqttException(MqttException.REASON_CODE_SUBSCRIBE_FAILED);
-		}
+		this.subscribe(topicFilters, qos, null);
 	}
 
 	/* (non-Javadoc)
@@ -463,17 +455,15 @@ public void subscribe(String topicFilter, int qos, IMqttMessageListener messageL
 	}
 
 
-	public void subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) throws MqttException {
-		this.subscribe(topicFilters, qos);
-
-		// add or remove message handlers to the list for this client
-		for (int i = 0; i < topicFilters.length; ++i) {
-            if (messageListeners[i] == null) {
-                aClient.comms.removeMessageListener(topicFilters[i]);
-            }
-            else {
-                aClient.comms.setMessageListener(topicFilters[i], messageListeners[i]);
-            }
+	public void subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) throws MqttException {	
+		IMqttToken tok = aClient.subscribe(topicFilters, qos, null, null, messageListeners);
+		tok.waitForCompletion(getTimeToWait());
+		int[] grantedQos = tok.getGrantedQos();
+		for (int i = 0; i < grantedQos.length; ++i) {
+			qos[i] = grantedQos[i];
+		}
+		if (grantedQos.length == 1 && qos[0] == 0x80) {
+			throw new MqttException(MqttException.REASON_CODE_SUBSCRIBE_FAILED);
 		}
 	}
 
@@ -533,22 +523,16 @@ public IMqttToken subscribeWithResponse(String[] topicFilters, IMqttMessageListe
 	 * @see IMqttClient#subscribeWithResponse(String[], int[])
 	 */
 	public IMqttToken subscribeWithResponse(String[] topicFilters, int[] qos) throws MqttException {
-		IMqttToken tok = aClient.subscribe(topicFilters, qos, null, null);
-		tok.waitForCompletion(getTimeToWait());
-		return tok;
+		return this.subscribeWithResponse(topicFilters, qos, null);
 	}
 
 	/*
 	 * @see IMqttClient#subscribeWithResponse(String[], int[], IMqttMessageListener[])
 	 */
 	public IMqttToken subscribeWithResponse(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners)
-			throws MqttException {
-		IMqttToken tok = this.subscribeWithResponse(topicFilters, qos);
-
-		// add message handlers to the list for this client
-		for (int i = 0; i < topicFilters.length; ++i) {
-			aClient.comms.setMessageListener(topicFilters[i], messageListeners[i]);
-		}
+			throws MqttException {		
+		IMqttToken tok = aClient.subscribe(topicFilters, qos, null, null, messageListeners);
+		tok.waitForCompletion(getTimeToWait());
 		return tok;
 	}