Fixes #421, #477, #509: QoS 0 tokens get lost.

QoS 0 messages have no message id, and the tokens of these messages
are thus all stored in the token store under id 0. When multiple
messages are posted in quick succession, the token of the previous
message is overwritten in the token store. As a result, clients waiting
on this token hang, and the inFlight message count is never decreased.

This commit attaches the token to the message itself. CommsSender and
ClientState first look in the message for a token, and then, if there is
no token there, in the token store. Tokens of QoS 0 messages are never
stored in the token store.

Signed-off-by: Hylke van der Schaaf <hylke.vds@gmail.com>

Author: hylkevds

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

@@ -502,7 +502,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) {
@@ -526,13 +527,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();
 			}
@@ -898,8 +900,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

@@ -131,7 +131,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
@@ -386,6 +395,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];
 	}