Issue #531 - Reverting misguided change that didn't assign Message IDs to QoS 0 messages

Signed-off-by: James Sutton <james.sutton@uk.ibm.com>

Author: jpwsutton

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

@@ -125,7 +125,7 @@ public class ClientState implements MqttState {
 	private long keepAlive;
 	private boolean cleanSession;
 	private MqttClientPersistence persistence;
-	
+
 	private int actualInFlight = 0;
 	private int inFlightPubRels = 0;
 
@@ -238,14 +238,14 @@ protected void clearState() throws MqttException {
 		outgoingTopicAliases.clear();
 		incomingTopicAliases.clear();
 	}
-	
+
 	protected void clearConnectionState() throws MqttException {
 		final String methodName = "clearConnectionState";
 		// @TRACE=665=Clearing Connection State (Topic Aliases)
 		log.fine(CLASS_NAME, methodName, "665");
 		outgoingTopicAliases.clear();
 		incomingTopicAliases.clear();
-		
+
 	}
 
 	private MqttWireMessage restoreMessage(String key, MqttPersistable persistable) throws MqttException {
@@ -520,14 +520,7 @@ public void send(MqttWireMessage message, MqttToken token) throws MqttException
 		final String methodName = "send";
 		// Set Message ID if required
 		if (message.isMessageIdRequired() && (message.getMessageId() == 0)) {
-			if (message instanceof MqttPublish && (((MqttPublish) message).getMessage().getQos() != 0)) {
-				message.setMessageId(getNextMessageId());
-			} else if (message instanceof MqttPubAck || message instanceof MqttPubRec || message instanceof MqttPubRel
-					|| message instanceof MqttPubComp || message instanceof MqttSubscribe
-					|| message instanceof MqttSubAck || message instanceof MqttUnsubscribe
-					|| message instanceof MqttUnsubAck) {
-				message.setMessageId(getNextMessageId());
-			}
+			message.setMessageId(getNextMessageId());
 		}
 		// Set Topic Alias if required
 		if (message instanceof MqttPublish && this.mqttSession.getOutgoingTopicAliasMaximum() > 0) {
@@ -545,7 +538,7 @@ public void send(MqttWireMessage message, MqttToken token) throws MqttException
 				}
 			}
 		}
-		
+
 		if (token != null) {
 			try {
 				token.internalTok.setMessageID(message.getMessageId());
@@ -1043,7 +1036,7 @@ protected void notifyReceivedAck(MqttAck ack) throws MqttException {
 			// @TRACE 662=no message found for ack id={0}
 			log.fine(CLASS_NAME, methodName, "662", new Object[] { Integer.valueOf(ack.getMessageId()) });
 		} else if (ack instanceof MqttPubRec) {
-			
+
 			// Update the token with the reason codes
 			updateResult(ack, token, mex);
 
@@ -1261,12 +1254,17 @@ protected void notifyComplete(MqttToken token) throws MqttException {
 			checkQuiesceLock();
 		}
 	}
-	
+
 	/**
-	 * Updates a token with the latest reason codes, currently only used for PubRec messages.
-	 * @param msg -  The message that we are using for the update
-	 * @param token - The Token we are updating
-	 * @param ex - if there was a problem store the exception in the token.
+	 * Updates a token with the latest reason codes, currently only used for PubRec
+	 * messages.
+	 * 
+	 * @param msg
+	 *            - The message that we are using for the update
+	 * @param token
+	 *            - The Token we are updating
+	 * @param ex
+	 *            - if there was a problem store the exception in the token.
 	 */
 	protected void updateResult(MqttWireMessage ack, MqttToken token, MqttException ex) {
 		token.internalTok.update(ack, ex);
@@ -1369,7 +1367,7 @@ public void disconnected(MqttException reason) {
 			if (cleanSession) {
 				clearState();
 			}
-			
+
 			clearConnectionState();
 
 			pendingMessages.clear();

org.eclipse.paho.mqttv5.client/src/test/java/org/eclipse/paho/mqttv5/client/test/PublishTests.java

@@ -88,6 +88,58 @@ public void testPublishRC() throws Exception {
 		int[] expectedRC = new int[] {16, 0};
 		Assert.assertArrayEquals(expectedRC, deliveryToken.getReasonCodes());
 
+		log.info("Disconnecting...");
+		IMqttToken disconnectToken = asyncClient.disconnect();
+		disconnectToken.waitForCompletion(5000);
+		Assert.assertFalse(asyncClient.isConnected());
+		asyncClient.close();
+
+	}
+	
+	@Test
+	public void testPublishManyQoS0Messages() throws Exception {
+		String methodName = Utility.getMethodName();
+		LoggingUtilities.banner(log, cclass, methodName);
+		String clientId = methodName;
+		MqttAsyncClient asyncClient = new MqttAsyncClient(serverURI.toString(), clientId);
+
+		// Connect to the server
+		log.info("Connecting: [serverURI: " + serverURI + ", ClientId: " + clientId + "]");
+		IMqttToken connectToken = asyncClient.connect();
+		connectToken.waitForCompletion(5000);
+		String clientId2 = asyncClient.getClientId();
+		log.info("Client ID = " + clientId2);
+		boolean isConnected = asyncClient.isConnected();
+		log.info("isConnected: " + isConnected);
+		
+		MqttMessage testMessage = new MqttMessage("Test Payload".getBytes(), 0, false, new MqttProperties());
+		long lStartTime = System.nanoTime();
+		int messagesSentThisSecond = 0;
+		long lastExTime = lStartTime;
+		for(int i = 0; i < 70000; i++) {
+			IMqttDeliveryToken deliveryToken = asyncClient.publish(topicPrefix + methodName, testMessage);
+			deliveryToken.waitForCompletion(1000);
+			messagesSentThisSecond++;
+			long now = System.nanoTime();
+			if(((now - lastExTime) /1000000) > 1000) {
+				lastExTime = now;
+				
+				System.out.println("Have sent " + messagesSentThisSecond + " messages in the last second.");
+				messagesSentThisSecond = 0;
+			}
+			
+		}
+		
+		//end
+        long lEndTime = System.nanoTime();
+
+		//time elapsed
+        long output = lEndTime - lStartTime;
+
+        System.out.println("Sending lots of messages  took : " + output / 1000000 + " milliseconds.");
+
+	
+
 		log.info("Disconnecting...");
 		IMqttToken disconnectToken = asyncClient.disconnect();
 		disconnectToken.waitForCompletion(5000);