Fix for max inflight overrun in V5 buffered reconnect #582

Author: Ian Craggs

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/ClientComms.java

@@ -173,6 +173,7 @@ void internalSend(MqttWireMessage message, MqttToken token) throws MqttException
 			// Persist if needed and send the message
 			this.clientState.send(message, token);
 		} catch (MqttException e) {
+			token.internalTok.setClient(null); // undo client setting on error
 			if (message instanceof MqttPublish) {
 				this.clientState.undo((MqttPublish) message);
 			}
@@ -946,12 +947,6 @@ class ReconnectDisconnectedBufferCallback implements IDisconnectedBufferCallback
 
 		public void publishBufferedMessage(BufferedMessage bufferedMessage) throws MqttException {
 			if (isConnected()) {
-				//int qos = ((MqttPublish) bufferedMessage.getMessage()).getQos();
-				while (clientState.getActualInFlight() >= (mqttConnection.getReceiveMaximum() - 4)) {
-					// We need to Yield to the other threads to allow the in flight messages to
-					// clear
-					Thread.yield();
-				}
 				// @TRACE 510=Publising Buffered message message={0}
 				log.fine(CLASS_NAME, methodName, "510", new Object[] { bufferedMessage.getMessage().getKey() });
 				internalSend(bufferedMessage.getMessage(), bufferedMessage.getToken());

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

@@ -512,7 +512,8 @@ public void send(MqttWireMessage message, MqttToken token) throws MqttException
 			message.setMessageId(getNextMessageId());
 		}
 		// Set Topic Alias if required
-		if (message instanceof MqttPublish && this.mqttConnection.getOutgoingTopicAliasMaximum() > 0) {
+		if (message instanceof MqttPublish && ((MqttPublish) message).getTopicName() != null 
+				&& this.mqttConnection.getOutgoingTopicAliasMaximum() > 0) {
 			String topic = ((MqttPublish) message).getTopicName();
 			if (outgoingTopicAliases.containsKey(topic)) {
 				// Existing Topic Alias, Assign it and remove the topic string

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/DisconnectedMessageBuffer.java

@@ -105,15 +105,15 @@ public void run() {
 		log.fine(CLASS_NAME, methodName, "516");
 			while(getMessageCount() > 0){
 				try {
-				BufferedMessage bufferedMessage = getMessage(0);
-				callback.publishBufferedMessage(bufferedMessage);
-				// Publish was successful, remove message from buffer.
-				deleteMessage(0);
+					BufferedMessage bufferedMessage = getMessage(0);
+					callback.publishBufferedMessage(bufferedMessage);
+					// Publish was successful, remove message from buffer.
+					deleteMessage(0);
 				} catch (MqttException ex) {
 					// Error occurred attempting to publish buffered message likely because the client is not connected
 					// @TRACE 519=Error occurred attempting to publish buffered message due to disconnect. Exception: {0}.
-					log.warning(CLASS_NAME, methodName, "519", new Object[]{ex.getMessage()});
-					break;
+					//log.warning(CLASS_NAME, methodName, "519", new Object[]{ex.getMessage()});
+					Thread.yield();
 				}
 			}
 	}