Issue #548 - Fixing duplicate QoS 2 issue

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

@@ -1167,6 +1167,8 @@ protected void handleInboundPubRel(MqttPubRel pubRel) throws MqttException {
 			// 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());
 			this.send(pubComp, null);
 		}
 	}

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

@@ -453,12 +453,16 @@ private void handleMessage(MqttPublish publishMessage) throws Exception {
 				this.clientComms.internalSend(
 						new MqttPubAck(MqttReturnCode.RETURN_CODE_SUCCESS, publishMessage.getMessageId(), new MqttProperties()),
 						new MqttToken(clientComms.getClient().getClientId()));
-			} else if (publishMessage.getMessage().getQos() == 2) {
+			} 
+			
+			/*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()));
-			}
+			}*/
 		}
 	}
 
@@ -469,6 +473,8 @@ public void messageArrivedComplete(int messageId, int qos) throws MqttException
 		} 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());
+
 			this.clientComms.internalSend(pubComp, new MqttToken(clientComms.getClient().getClientId()));
 		}
 	}

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

@@ -121,8 +121,8 @@ public MqttWireMessage readMqttWireMessage() throws IOException, MqttException {
 				byte[] header = bais.toByteArray();
 				System.arraycopy(header,0,packet,0, header.length);
 				message = MqttWireMessage.createWireMessage(packet);
-				// @TRACE 501= received {0} 
-				log.fine(CLASS_NAME, methodName, "501",new Object[] {message});
+				// @TRACE 530= Received {0} 
+				log.fine(CLASS_NAME, methodName, "530",new Object[] {message});
 			}
 		} catch (SocketTimeoutException e) {
 			// ignore socket read timeout

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

@@ -81,6 +81,7 @@
 517=Un-Persisting Buffered message key={0}
 518=Failed to Un-Persist Buffered message key={0}
 529=Sent {0}
+530=Received {0}
 600=>
 601=key={0} message={1}
 602=key={0} exception

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

@@ -52,8 +52,6 @@ public static void setUpBeforeClass() throws Exception {
 		}
 	}
 
-	
-
 	/**
 	 * Very simple test case that validates that the client can connect and then
 	 * disconnect cleanly.
@@ -118,4 +116,58 @@ public void testPublishAndReceive() throws MqttException, InterruptedException {
 		TestClientUtilities.disconnectAndCloseClient(asyncClient, timeout);
 	}
 
+	/**
+	 * @throws MqttException,
+	 *             InterruptedException
+	 * 
+	 */
+	@Test
+	public void testQoS2DuplicateIssue() throws MqttException, InterruptedException {
+		String methodName = Utility.getMethodName();
+		LoggingUtilities.banner(log, SubscribeTests.class, methodName);
+		String subTopic = "a/+/c";
+
+		int timeout = 5000;
+
+		MqttV5Receiver mqttV5Receiver = new MqttV5Receiver(methodName, LoggingUtilities.getPrintStream());
+		MqttAsyncClient asyncClient = TestClientUtilities.connectAndGetClient(serverURI.toString(), methodName,
+				mqttV5Receiver, null, timeout);
+
+		int qos = 2;
+		log.info("Testing Publish and Receive at QoS: " + qos);
+		// Subscribe to a topic
+		log.info(String.format("Subscribing to: %s at QoS %d", subTopic, qos));
+		MqttSubscription subscription = new MqttSubscription(subTopic, qos);
+		IMqttToken subscribeToken = asyncClient.subscribe(subscription);
+		subscribeToken.waitForCompletion(timeout);
+		
+		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()));
+		
+		Thread.sleep(10000);
+
+
+		TestClientUtilities.disconnectAndCloseClient(asyncClient, timeout);
+	}
+
+	private void publishMessage(String payload, int qos, String topic, MqttAsyncClient client, int timeout) throws MqttException {
+		MqttMessage testMessage = new MqttMessage(payload.getBytes(), qos, false, null);
+		log.info(String.format("Publishing Message %s to: %s at QoS: %d", testMessage.toDebugString(), topic,
+				qos));
+		IMqttDeliveryToken deliveryToken = client.publish(topic, testMessage);
+		deliveryToken.waitForCompletion(timeout);
+	}
+
 }