Issue #548 - Further tweaks that should fix Dup PubCom on QoS2 as well as dup message delivery.

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

Author: jpwsutton

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/TimerPingSender.java

@@ -35,17 +35,19 @@ public class TimerPingSender implements MqttPingSender {
 
 	private ClientComms comms;
 	private Timer timer;
+	private String clientid;
 
 	public void init(ClientComms comms) {
 		if (comms == null) {
 			throw new IllegalArgumentException("ClientComms cannot be null.");
 		}
 		this.comms = comms;
+		clientid = comms.getClient().getClientId();
+		log.setResourceName(clientid);
 	}
 
 	public void start() {
 		final String methodName = "start";		
-		String clientid = comms.getClient().getClientId();
 
 		//@Trace 659=start timer for client:{0}
 		log.fine(CLASS_NAME, methodName, "659", new Object[]{clientid});

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/TimerPingSender.java

@@ -53,6 +53,7 @@ public void init(ClientComms comms) {
 		}
 		this.comms = comms;
 		clientid = comms.getClient().getClientId();
+		log.setResourceName(clientid);
 	}
 
 	public void start() {

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

@@ -1150,25 +1150,18 @@ protected void handleOrphanedAcks(MqttAck ack) throws MqttException {
 	 */
 	protected void handleInboundPubRel(MqttPubRel pubRel) throws MqttException {
 		final String methodName = "handleInboundPubRel";
-		MqttPublish sendMsg = (MqttPublish) inboundQoS2.get(Integer.valueOf(pubRel.getMessageId()));
 		if (pubRel.getReasonCodes()[0] > MqttReturnCode.RETURN_CODE_UNSPECIFIED_ERROR) {
 			// @TRACE 667=MqttPubRel was received with an error code: key={0} message={1},
 			// Reason Code={2}
 			log.severe(CLASS_NAME, methodName, "667",
 					new Object[] { pubRel.getMessageId(), pubRel.toString(), pubRel.getReasonCodes()[0] });
 			throw new MqttException(pubRel.getReasonCodes()[0]);
-		}
-		if (sendMsg != null) {
-			if (callback != null) {
-				callback.messageArrived(sendMsg);
-			}
 		} else {
-			// Original publish has already been delivered.
 			// Currently this client has no need of the properties, so this is left empty.
 			MqttPubComp pubComp = new MqttPubComp(MqttReturnCode.RETURN_CODE_SUCCESS, pubRel.getMessageId(),
 					new MqttProperties());
-			// TODO - Fix Trace
-			log.info(CLASS_NAME, methodName, "Creating MqttPubComp due to pubRel: " + pubComp.toString());
+			// @TRACE 668=Creating MqttPubComp: {0}
+			log.info(CLASS_NAME, methodName, "668", new Object[] { pubComp.toString()});
 			this.send(pubComp, null);
 		}
 	}

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

@@ -302,7 +302,8 @@ public void connectionLost(MqttException cause, MqttDisconnect message) {
 				// @TRACE 722=Server initiated disconnect, connection closed. Disconnect={0}
 				log.fine(CLASS_NAME, methodName, "722", new Object[] { message.toString() });
 				MqttDisconnectResponse disconnectResponse = new MqttDisconnectResponse(message.getReturnCode(),
-						message.getProperties().getReasonString(), (ArrayList<UserProperty>) message.getProperties().getUserProperties(),
+						message.getProperties().getReasonString(),
+						(ArrayList<UserProperty>) message.getProperties().getUserProperties(),
 						message.getProperties().getServerReference());
 				mqttCallback.disconnected(disconnectResponse);
 			} else if (mqttCallback != null && cause != null) {
@@ -385,13 +386,15 @@ public void messageArrived(MqttPublish sendMessage) {
 			}
 		}
 	}
-	
+
 	/**
 	 * This method is called when an Auth Message is received.
-	 * @param authMessage The {@link MqttAuth} message.
+	 * 
+	 * @param authMessage
+	 *            The {@link MqttAuth} message.
 	 */
 	public void authMessageReceived(MqttAuth authMessage) {
-		if(mqttCallback != null) {
+		if (mqttCallback != null) {
 			mqttCallback.authPacketArrived(authMessage.getReturnCode(), authMessage.getProperties());
 		}
 	}
@@ -427,11 +430,11 @@ public void quiesce() {
 			spaceAvailable.notifyAll();
 		}
 	}
-	
+
 	boolean areQueuesEmpty() {
-	    synchronized (workAvailable) {
-	        return completeQueue.isEmpty() && messageQueue.isEmpty();    
-	    }
+		synchronized (workAvailable) {
+			return completeQueue.isEmpty() && messageQueue.isEmpty();
+		}
 	}
 
 	public boolean isQuiesced() {
@@ -441,39 +444,30 @@ public boolean isQuiesced() {
 	private void handleMessage(MqttPublish publishMessage) throws Exception {
 		final String methodName = "handleMessage";
 		// If quisecing process any pending messages.
-
 		String destName = publishMessage.getTopicName();
 
 		// @TRACE 713=call messageArrived key={0} topic={1}
 		log.fine(CLASS_NAME, methodName, "713", new Object[] { new Integer(publishMessage.getMessageId()), destName });
 		deliverMessage(destName, publishMessage.getMessageId(), publishMessage.getMessage());
 
-		if (!this.manualAcks) {
-			if (publishMessage.getMessage().getQos() == 1) {
-				this.clientComms.internalSend(
-						new MqttPubAck(MqttReturnCode.RETURN_CODE_SUCCESS, publishMessage.getMessageId(), new MqttProperties()),
-						new MqttToken(clientComms.getClient().getClientId()));
-			} 
-			
-			/*else if (publishMessage.getMessage().getQos() == 2) {
-				this.clientComms.deliveryComplete(publishMessage);
-				MqttPubComp pubComp = new MqttPubComp(MqttReturnCode.RETURN_CODE_SUCCESS,
-						publishMessage.getMessageId(), new MqttProperties());
-				log.info(CLASS_NAME, methodName, "Creating MqttPubComp due to handleMessage: " + pubComp.toString());
-
-				this.clientComms.internalSend(pubComp, new MqttToken(clientComms.getClient().getClientId()));
-			}*/
+		// If we are not in manual ACK mode:
+		if (!this.manualAcks && publishMessage.getMessage().getQos() == 1) {
+			this.clientComms.internalSend(new MqttPubAck(MqttReturnCode.RETURN_CODE_SUCCESS,
+					publishMessage.getMessageId(), new MqttProperties()),
+					new MqttToken(clientComms.getClient().getClientId()));
 		}
 	}
 
 	public void messageArrivedComplete(int messageId, int qos) throws MqttException {
 		if (qos == 1) {
-			this.clientComms.internalSend(new MqttPubAck(MqttReturnCode.RETURN_CODE_SUCCESS, messageId, new MqttProperties()),
+			this.clientComms.internalSend(
+					new MqttPubAck(MqttReturnCode.RETURN_CODE_SUCCESS, messageId, new MqttProperties()),
 					new MqttToken(clientComms.getClient().getClientId()));
 		} else if (qos == 2) {
 			this.clientComms.deliveryComplete(messageId);
 			MqttPubComp pubComp = new MqttPubComp(MqttReturnCode.RETURN_CODE_SUCCESS, messageId, new MqttProperties());
-			log.info(CLASS_NAME, "messageArrivedComplete", "Creating MqttPubComp due to messageArrivedComplete: " + pubComp.toString());
+			// @TRACE 723=Creating MqttPubComp due to manual ACK: {0}
+			log.info(CLASS_NAME, "messageArrivedComplete", "723", new Object[] {pubComp.toString()});
 
 			this.clientComms.internalSend(pubComp, new MqttToken(clientComms.getClient().getClientId()));
 		}

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

@@ -51,7 +51,7 @@ public class CommsReceiver implements Runnable {
 	private Future<?> receiverFuture;
 
 	public CommsReceiver(ClientComms clientComms, ClientState clientState, CommsTokenStore tokenStore, InputStream in) {
-		this.in = new MqttInputStream(clientState, in);
+		this.in = new MqttInputStream(clientState, in, clientComms.getClient().getClientId());
 		this.clientComms = clientComms;
 		this.clientState = clientState;
 		this.tokenStore = tokenStore;

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

@@ -51,7 +51,7 @@ public class CommsSender implements Runnable {
 	private Future<?> senderFuture;
 
 	public CommsSender(ClientComms clientComms, ClientState clientState, CommsTokenStore tokenStore, OutputStream out) {
-		this.out = new MqttOutputStream(clientState, out);
+		this.out = new MqttOutputStream(clientState, out, clientComms.getClient().getClientId());
 		this.clientComms = clientComms;
 		this.clientState = clientState;
 		this.tokenStore = tokenStore;

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/wire/MqttInputStream.java

@@ -47,11 +47,12 @@ public class MqttInputStream extends InputStream {
 	private int packetLen;
 	private byte[] packet;
 
-	public MqttInputStream(MqttState clientState, InputStream in) {
+	public MqttInputStream(MqttState clientState, InputStream in, String clientId) {
 		this.clientState = clientState;
 		this.in = new DataInputStream(in);		
 		this.bais = new ByteArrayOutputStream();
 		this.remLen = -1;
+		log.setResourceName(clientId);
 	}
 
 	public int read() throws IOException {

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/wire/MqttOutputStream.java

@@ -37,9 +37,10 @@ public class MqttOutputStream extends OutputStream {
 	private MqttState clientState = null;
 	private BufferedOutputStream out;
 
-	public MqttOutputStream(MqttState clientState, OutputStream out) {
+	public MqttOutputStream(MqttState clientState, OutputStream out, String clientId) {
 		this.clientState = clientState;
 		this.out = new BufferedOutputStream(out);
+		log.setResourceName(clientId);
 	}
 
 	public void close() throws IOException {

org.eclipse.paho.mqttv5.client/src/main/resources/org/eclipse/paho/mqttv5/client/internal/nls/logcat.properties

@@ -145,6 +145,7 @@
 665=Clearing Connection State (Topic Aliases)
 666=Orphaned Ack key={0} message={1}
 667=MqttPubRel was received with an error code: key={0} message={1}, Reason Code= {2}
+668=Creating MqttPubComp: {0}
 700=stopping
 701=notify workAvailable and wait for run
 703=stopped
@@ -164,6 +165,7 @@
 720=exception from connectionLost {0}
 721=Non-Critical MQTT error thrown, passing back to application={0}
 722=Server initiated disconnect, connection closed. Disconnect={0}
+723=Creating MqttPubComp due to manual ACK: {0}
 800=stopping sender
 801=stopped
 802=network send key={0} msg={1}

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

@@ -8,6 +8,8 @@
 import org.eclipse.paho.mqttv5.client.IMqttDeliveryToken;
 import org.eclipse.paho.mqttv5.client.IMqttToken;
 import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
+import org.eclipse.paho.mqttv5.client.MqttCallback;
+import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
 import org.eclipse.paho.mqttv5.client.test.logging.LoggingUtilities;
 import org.eclipse.paho.mqttv5.client.test.properties.TestProperties;
 import org.eclipse.paho.mqttv5.client.test.utilities.MqttV5Receiver;
@@ -16,6 +18,7 @@
 import org.eclipse.paho.mqttv5.common.MqttException;
 import org.eclipse.paho.mqttv5.common.MqttMessage;
 import org.eclipse.paho.mqttv5.common.MqttSubscription;
+import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -144,20 +147,14 @@ public void testQoS2DuplicateIssue() throws MqttException, InterruptedException
 		String samplePayload = "Hello World";
 
 		log.info("Publishing messages on topics.");
-		//publishMessage(samplePayload, 0, "a/b/c", asyncClient, timeout);
-		//publishMessage(samplePayload, 0, "a/0/c", asyncClient, timeout);
-		//publishMessage(samplePayload, 1, "a/1/c", asyncClient, timeout);
 		publishMessage(samplePayload, 2, "a/2/c", asyncClient, timeout);
 
 
 		log.info("Waiting for delivery and validating message.");
-		//Assert.assertTrue(mqttV5Receiver.validateReceipt("a/b/c", 0, samplePayload.getBytes()));
-		//Assert.assertTrue(mqttV5Receiver.validateReceipt("a/0/c", 0, samplePayload.getBytes()));
-		//Assert.assertTrue(mqttV5Receiver.validateReceipt("a/1/c", 1, samplePayload.getBytes()));
 		Assert.assertTrue(mqttV5Receiver.validateReceipt("a/2/c", 2, samplePayload.getBytes()));
+		log.info("Number of received message: " + mqttV5Receiver.receivedMessageCount());
+				
 
-		Thread.sleep(10000);
-
 
 		TestClientUtilities.disconnectAndCloseClient(asyncClient, timeout);
 	}