Fix for issue #432 - not receiving retained messages

Author: Ian Craggs

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

@@ -925,7 +925,6 @@ public IMqttToken subscribe(String[] topicFilters, int[] qos) throws MqttExcepti
 	 */
 	public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext, IMqttActionListener callback)
 			throws MqttException {
-		final String methodName = "subscribe";
 
 		if (topicFilters.length != qos.length) {
 			throw new IllegalArgumentException();
@@ -937,17 +936,22 @@ public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext
 			MqttTopic.validate(topicFilters[i], true/* allow wildcards */);
 			this.comms.removeMessageListener(topicFilters[i]);
 		}
+		
+		return this.subscribeBase(topicFilters, qos, userContext, callback);
+	}
 
+	private IMqttToken subscribeBase(String[] topicFilters, int[] qos, Object userContext, IMqttActionListener callback)
+			throws MqttException {
+		final String methodName = "subscribe";
+				
 		// Only Generate Log string if we are logging at FINE level
 		if (log.isLoggable(Logger.FINE)) {
 			StringBuffer subs = new StringBuffer();
 			for (int i = 0; i < topicFilters.length; i++) {
 				if (i > 0) {
 					subs.append(", ");
 				}
-				subs.append("topic=").append(topicFilters[i]).append(" qos=").append(qos[i]);
-
-				
+				subs.append("topic=").append(topicFilters[i]).append(" qos=").append(qos[i]);			
 			}
 			// @TRACE 106=Subscribe topicFilter={0} userContext={1} callback={2}
 			log.fine(CLASS_NAME, methodName, "106", new Object[] { subs.toString(), userContext, callback });
@@ -1010,20 +1014,19 @@ public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext
 		if ((messageListeners.length != qos.length) || (qos.length != topicFilters.length)) {
 			throw new IllegalArgumentException();
 		}
-
-		IMqttToken token = this.subscribe(topicFilters, qos, userContext, callback);
-
+		
 		// 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) {
                 this.comms.removeMessageListener(topicFilters[i]);
             }
             else {
                 this.comms.setMessageListener(topicFilters[i], messageListeners[i]);
             }
 		}
-
-		return token;
+		
+		return this.subscribeBase(topicFilters, qos, userContext, callback);
 	}
 
 	/*

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

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2016 IBM Corp.
+ * Copyright (c) 2009, 2018 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -11,12 +11,7 @@
  *   http://www.eclipse.org/org/documents/edl-v10.php.
  *
  * Contributors:
- *    Dave Locke - initial API and implementation and/or initial documentation
- *    Ian Craggs - MQTT 3.1.1 support
- *    Ian Craggs - per subscription message handlers (bug 466579)
- *    Ian Craggs - ack control (bug 472172)
- *    James Sutton - Bug 459142 - WebSocket support for the Java client.
- *    James Sutton - Automatic Reconnect & Offline Buffering.
+ *    James Sutton - initial API and implementation and/or initial documentation
  */
 
 package org.eclipse.paho.mqttv5.client;
@@ -1107,7 +1102,6 @@ public IMqttToken subscribe(MqttSubscription[] subscriptions) throws MqttExcepti
 	@Override
 	public IMqttToken subscribe(MqttSubscription[] subscriptions, Object userContext, MqttActionListener callback,
 			MqttProperties subscriptionProperties) throws MqttException {
-		final String methodName = "subscribe";
 
 		// remove any message handlers for individual topics and validate Topics
 		for (int i = 0; i < subscriptions.length; ++i) {
@@ -1117,7 +1111,14 @@ public IMqttToken subscribe(MqttSubscription[] subscriptions, Object userContext
 					this.mqttConnection.isWildcardSubscriptionsAvailable(),
 					this.mqttConnection.isSharedSubscriptionsAvailable());
 		}
+		
+		return this.subscribeBase(subscriptions, userContext, callback, subscriptionProperties);
+	}
 
+	private IMqttToken subscribeBase(MqttSubscription[] subscriptions, Object userContext, MqttActionListener callback,
+				MqttProperties subscriptionProperties) throws MqttException {
+		final String methodName = "subscribe";
+		
 		// Only Generate Log string if we are logging at FINE level
 		if (log.isLoggable(Logger.FINE)) {
 			StringBuffer subs = new StringBuffer();
@@ -1208,14 +1209,15 @@ public IMqttToken subscribe(MqttSubscription[] subscriptions, IMqttMessageListen
 	public IMqttToken subscribe(MqttSubscription[] subscriptions, Object userContext, MqttActionListener callback,
 			IMqttMessageListener[] messageListeners, MqttProperties subscriptionProperties) throws MqttException {
 
-		IMqttToken token = this.subscribe(subscriptions, userContext, callback, subscriptionProperties);
-
 		// add message handlers to the list for this client
 		for (int i = 0; i < subscriptions.length; ++i) {
+			MqttTopicValidator.validate(subscriptions[i].getTopic(),
+					this.mqttConnection.isWildcardSubscriptionsAvailable(),
+					this.mqttConnection.isSharedSubscriptionsAvailable());
 			this.comms.setMessageListener(null, subscriptions[i].getTopic(), messageListeners[i]);
 		}
-
-		return token;
+		
+		return this.subscribeBase(subscriptions, userContext, callback, subscriptionProperties);
 	}
 
 	/*
@@ -1251,15 +1253,16 @@ public IMqttToken subscribe(MqttSubscription[] subscriptions, Object userContext
 				subId = this.mqttSession.getNextSubscriptionIdentifier();
 			}
 		}
-
-		IMqttToken token = this.subscribe(subscriptions, userContext, callback, subscriptionProperties);
-
+		
 		// add message handlers to the list for this client
 		for (int i = 0; i < subscriptions.length; ++i) {
+			MqttTopicValidator.validate(subscriptions[i].getTopic(),
+					this.mqttConnection.isWildcardSubscriptionsAvailable(),
+					this.mqttConnection.isSharedSubscriptionsAvailable());
 			this.comms.setMessageListener(subId, subscriptions[i].getTopic(), messageListener);
 		}
 
-		return token;
+		return this.subscribeBase(subscriptions, userContext, callback, subscriptionProperties);
 	}
 
 	/*