Fix max_inflight condition for disconnected buffer #582

Author: Ian Craggs

org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/automaticReconnect/OfflineBufferingTest.java

@@ -173,7 +173,7 @@ public void testManyMessageBufferAndDeliver() throws Exception {
 		options.setAutomaticReconnect(true);
 
 		// Workaround for Issue #582 - Remove once fixed.
-		options.setMaxInflight(100);
+		//options.setMaxInflight(100);
 
 		MqttAsyncClient client = new MqttAsyncClient("tcp://localhost:" + proxy.getLocalPort(), methodName, DATA_STORE);
 		DisconnectedBufferOptions disconnectedOpts = new DisconnectedBufferOptions();
@@ -203,9 +203,11 @@ public void testManyMessageBufferAndDeliver() throws Exception {
 		isConnected = client.isConnected();
 		log.info("Proxy Disconnect isConnected: " + isConnected);
 		Assert.assertFalse(isConnected);
+		
+		int msg_count = 100;
 
-		// Publish 100 messages
-		for (int x = 0; x < 100; x++) {
+		// Publish msg_count messages
+		for (int x = 0; x < msg_count; x++) {
 			client.publish(topicPrefix + methodName, new MqttMessage(Integer.toString(x).getBytes()));
 		}
 		// Enable Proxy
@@ -234,7 +236,7 @@ public void testManyMessageBufferAndDeliver() throws Exception {
 		Thread.sleep(5000);
 
 		// Check that all messages have been delivered
-		for (int x = 0; x < 100; x++) {
+		for (int x = 0; x < msg_count; x++) {
 			boolean recieved = mqttV3Receiver.validateReceipt(topicPrefix + methodName, 1,
 					Integer.toString(x).getBytes());
 			Assert.assertTrue(recieved);

org.eclipse.paho.client.mqttv3/src/main/java-templates/org/eclipse/paho/client/mqttv3/internal/ClientComms.java

@@ -160,6 +160,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);
 			}

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

@@ -102,6 +102,7 @@ public int getMessageCount() {
 		}
 	}
 
+	private int mycount = 0;
 	/**
 	 * Flushes the buffer of messages into an open connection
 	 */
@@ -116,12 +117,17 @@ public void run() {
 				// 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}:{1}.
-				log.severe(CLASS_NAME, methodName, "519", new Object[] { ex.getReasonCode(), ex.getMessage() });
-				break;
-
+				if (ex.getReasonCode() == MqttException.REASON_CODE_MAX_INFLIGHT) {
+					// If we get the max_inflight condition, try again after a short
+					// interval to allow more messages to be completely sent.
+					try { Thread.sleep(100); } catch (Exception e) {}
+				} else {
+					// 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}:{1}.
+					log.severe(CLASS_NAME, methodName, "519", new Object[] { ex.getReasonCode(), ex.getMessage() });
+					break;
+				}
 			}
 
 		}

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

@@ -110,10 +110,16 @@ public void run() {
 					// 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()});
-					Thread.yield();
+					if (ex.getReasonCode() == MqttClientException.REASON_CODE_MAX_INFLIGHT) {
+						// If we get the max_inflight condition, try again after a short
+						// interval to allow more messages to be completely sent.
+						try { Thread.sleep(100); } catch (Exception e) {}
+					} else {
+						// 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;
+					}
 				}
 			}
 	}

org.eclipse.paho.mqttv5.client/src/test/java/org/eclipse/paho/mqttv5/client/test/automaticReconnect/OfflineBufferingTest.java

@@ -157,12 +157,16 @@ public void testManyMessageBufferAndDeliver() throws Exception {
 		// Tokens
 		IMqttToken connectToken;
 
+		int msg_count = 1000;
+		
 		// Client Options
 		MqttConnectionOptions options = new MqttConnectionOptions();
 		options.setCleanStart(true);
 		options.setAutomaticReconnect(true);
+		
 		MqttAsyncClient client = new MqttAsyncClient("tcp://localhost:" + proxy.getLocalPort(), methodName, DATA_STORE);
 		DisconnectedBufferOptions disconnectedOpts = new DisconnectedBufferOptions();
+		disconnectedOpts.setBufferSize(msg_count);
 		disconnectedOpts.setBufferEnabled(true);
 		client.setBufferOpts(disconnectedOpts);
 
@@ -189,8 +193,8 @@ public void testManyMessageBufferAndDeliver() throws Exception {
 		log.info("Proxy Disconnect isConnected: " + isConnected);
 		Assert.assertFalse(isConnected);
 
-		// Publish 100 messages
-		for (int x = 0; x < 100; x++) {
+		// Publish some messages
+		for (int x = 0; x < msg_count; x++) {
 			client.publish(topicPrefix + methodName, new MqttMessage(Integer.toString(x).getBytes()));
 		}
 		// Enable Proxy
@@ -219,11 +223,11 @@ public void testManyMessageBufferAndDeliver() throws Exception {
 		Thread.sleep(5000);
 
 		// Check that all messages have been delivered
-		for (int x = 0; x < 100; x++) {
-			boolean recieved = mqttV3Receiver.validateReceipt(topicPrefix + methodName, 0, Integer.toString(x).getBytes());
-			Assert.assertTrue(recieved);
+		for (int x = 0; x < msg_count; x++) {
+			boolean received = mqttV3Receiver.validateReceipt(topicPrefix + methodName, 0, Integer.toString(x).getBytes());
+			Assert.assertTrue(received);
 		}
-		log.info("All messages sent and Recieved correctly.");
+		log.info("All messages sent and received correctly.");
 		IMqttToken disconnectToken = client.disconnect();
 		disconnectToken.waitForCompletion(5000);
 		client.close();