Fix Bug 431268 - DUP flag incorrectly set by paho client

This issue is cause by caching the encoded message header, and didn't set DUP flag when retransmission in case of reconnect.

Bug: 431268
Signed-off-by: bzhangxmt <zhbinbj@cn.ibm.com>

Author: bzhangxmt

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

@@ -315,9 +315,11 @@ protected void restoreState() throws MqttException {
 					if (persistence.containsKey(getSendConfirmPersistenceKey(sendMessage))) {
 						MqttPersistable persistedConfirm = persistence.get(getSendConfirmPersistenceKey(sendMessage));
 						// QoS 2, and CONFIRM has already been sent...
+						// NO DUP flag is allowed for 3.1.1 spec while it's not clear for 3.1 spec
+						// So we just remove DUP
 						MqttPubRel confirmMessage = (MqttPubRel) restoreMessage(key, persistedConfirm);
 						if (confirmMessage != null) {
-							confirmMessage.setDuplicate(true);
+							// confirmMessage.setDuplicate(true); // REMOVED
 							//@TRACE 605=outbound QoS 2 pubrel key={0} message={1}
 							log.fine(CLASS_NAME,methodName, "605", new Object[]{key,message});
 
@@ -375,11 +377,12 @@ private void restoreInflightMessages() {
 		Enumeration keys = outboundQoS2.keys();
 		while (keys.hasMoreElements()) {
 			Object key = keys.nextElement();
-			Object msg = outboundQoS2.get(key);
+			MqttWireMessage msg = (MqttWireMessage) outboundQoS2.get(key);
 			if (msg instanceof MqttPublish) {
 				//@TRACE 610=QoS 2 publish key={0}
 				log.fine(CLASS_NAME,methodName, "610", new Object[]{key});
-
+                // set DUP flag only for PUBLISH, but NOT for PUBREL (spec 3.1.1)
+				msg.setDuplicate(true);  
 				insertInOrder(pendingMessages, (MqttPublish)msg);
 			} else if (msg instanceof MqttPubRel) {
 				//@TRACE 611=QoS 2 pubrel key={0}
@@ -392,6 +395,7 @@ private void restoreInflightMessages() {
 		while (keys.hasMoreElements()) {
 			Object key = keys.nextElement();
 			MqttPublish msg = (MqttPublish)outboundQoS1.get(key);
+			msg.setDuplicate(true);
 			//@TRACE 612=QoS 1 publish key={0}
 			log.fine(CLASS_NAME,methodName, "612", new Object[]{key});
 

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/wire/MqttWireMessage.java

@@ -60,7 +60,6 @@ public abstract class MqttWireMessage {
 
 	protected boolean duplicate = false;
 
-	private byte[] encodedHeader = null;
 
 	public MqttWireMessage(byte type) {
 		this.type = type;
@@ -114,24 +113,21 @@ public String getKey() {
 	}
 
 	public byte[] getHeader() throws MqttException {
-		if (encodedHeader == null) {
-			try {
-				int first = ((getType() & 0x0f) << 4) ^ (getMessageInfo() & 0x0f);
-				byte[] varHeader = getVariableHeader();
-				int remLen = varHeader.length + getPayload().length;
+	    try {
+	        int first = ((getType() & 0x0f) << 4) ^ (getMessageInfo() & 0x0f);
+	        byte[] varHeader = getVariableHeader();
+	        int remLen = varHeader.length + getPayload().length;
 
-				ByteArrayOutputStream baos = new ByteArrayOutputStream();
-				DataOutputStream dos = new DataOutputStream(baos);
-				dos.writeByte(first);
-				dos.write(encodeMBI(remLen));
-				dos.write(varHeader);
-				dos.flush();
-				encodedHeader = baos.toByteArray();
-			} catch(IOException ioe) {
-				throw new MqttException(ioe);
-			}
-		}
-		return encodedHeader;
+	        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+	        DataOutputStream dos = new DataOutputStream(baos);
+	        dos.writeByte(first);
+	        dos.write(encodeMBI(remLen));
+	        dos.write(varHeader);
+	        dos.flush();
+	        return baos.toByteArray();
+	    } catch(IOException ioe) {
+	        throw new MqttException(ioe);
+	    }
 	}
 
 	protected abstract byte[] getVariableHeader() throws MqttException;