Merge branch 'hylkevds-develop' into develop

Author: Ian Craggs

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/ClientState.java

@@ -507,7 +507,8 @@ public void send(MqttWireMessage message, MqttToken token) throws MqttException
 					message.setMessageId(getNextMessageId());
 				}
 		}
-		if (token != null ) {
+		if (token != null) {
+			message.setToken(token);
 			try {
 				token.internalTok.setMessageID(message.getMessageId());
 			} catch (Exception e) {
@@ -531,13 +532,14 @@ public void send(MqttWireMessage message, MqttToken token) throws MqttException
 					case 2:
 						outboundQoS2.put( Integer.valueOf(message.getMessageId()), message);
 						persistence.put(getSendPersistenceKey(message), (MqttPublish) message);
+						tokenStore.saveToken(token, message);
 						break;
 					case 1:
 						outboundQoS1.put( Integer.valueOf(message.getMessageId()), message);
 						persistence.put(getSendPersistenceKey(message), (MqttPublish) message);
+						tokenStore.saveToken(token, message);
 						break;
 				}
-				tokenStore.saveToken(token, message);
 				pendingMessages.addElement(message);
 				queueLock.notifyAll();
 			}
@@ -906,8 +908,11 @@ protected void notifySent(MqttWireMessage message) {
 		//@TRACE 625=key={0}
 		log.fine(CLASS_NAME,methodName,"625",new Object[]{message.getKey()});
 
-		MqttToken token = tokenStore.getToken(message);
-		if (token == null) return;
+		MqttToken token = message.getToken();
+		if (token == null) {
+			token = tokenStore.getToken(message);
+			if (token == null) return;
+		}
 		token.internalTok.notifySent();
         if (message instanceof MqttPingReq) {
             synchronized (pingOutstandingLock) {

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/CommsSender.java

@@ -134,7 +134,10 @@ public void run() {
 							out.write(message);
 							out.flush();
 						} else {
-							MqttToken token = tokenStore.getToken(message);
+							MqttToken token = message.getToken();
+							if (token == null) {
+								tokenStore.getToken(message);
+							}
 							// While quiescing the tokenstore can be cleared so need
 							// to check for null for the case where clear occurs
 							// while trying to send a message.

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/wire/MqttWireMessage.java

@@ -27,6 +27,7 @@
 
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttPersistable;
+import org.eclipse.paho.client.mqttv3.MqttToken;
 import org.eclipse.paho.client.mqttv3.internal.ExceptionHelper;
 
 /**
@@ -65,6 +66,14 @@ public abstract class MqttWireMessage {
 
 	protected boolean duplicate = false;
 
+	/**
+	 * The token associated with the message. It needs to be stored here,
+	 * because QoS 0 messages do not have an ID, and tokens for these messages
+	 * can thus not be stored in the Token Store.
+	 */
+	private MqttToken token;
+
+
 	public MqttWireMessage(byte type) {
 		this.type = type;
 		// Use zero as the default message ID. Can't use -1, as that is serialized
@@ -409,6 +418,24 @@ public static void validateVariableByteInt(int value) throws IllegalArgumentExce
 
 	}
 
+	/**
+	 * Get the token associated with the message.
+	 *
+	 * @return The token associated with the message.
+	 */
+	public MqttToken getToken() {
+		return token;
+	}
+
+	/**
+	 * Set the token associated with the message.
+	 *
+	 * @param token the token associated with the message.
+	 */
+	public void setToken(MqttToken token) {
+		this.token = token;
+	}
+
 	public String toString() {
 		return PACKET_NAMES[type];
 	}