Merge pull request #225 from jpwsutton/develop

Resolving main issues in Issue #224

Author: jpwsutton

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

@@ -150,7 +150,9 @@ public void sendNoWait(MqttWireMessage message, MqttToken token) throws MqttExce
 			if(disconnectedMessageBuffer != null && disconnectedMessageBuffer.getMessageCount() != 0){
 				//@TRACE 507=Client Connected, Offline Buffer available, but not empty. Adding message to buffer. message={0}
 				log.fine(CLASS_NAME, methodName, "507", new Object[] {message.getKey()});
-				this.clientState.persistBufferedMessage(message);
+				if(disconnectedMessageBuffer.isPersistBuffer()){
+					this.clientState.persistBufferedMessage(message);
+				}
 				disconnectedMessageBuffer.putMessage(message, token);
 			} else {
 				this.internalSend(message, token);

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

@@ -45,6 +45,7 @@
 import org.eclipse.paho.client.mqttv3.internal.wire.MqttPubRec;
 import org.eclipse.paho.client.mqttv3.internal.wire.MqttPubRel;
 import org.eclipse.paho.client.mqttv3.internal.wire.MqttPublish;
+import org.eclipse.paho.client.mqttv3.internal.wire.MqttSuback;
 import org.eclipse.paho.client.mqttv3.internal.wire.MqttWireMessage;
 import org.eclipse.paho.client.mqttv3.logging.Logger;
 import org.eclipse.paho.client.mqttv3.logging.LoggerFactory;
@@ -562,6 +563,7 @@ public void persistBufferedMessage(MqttWireMessage message) {
 		// Because the client will have disconnected, we will want to re-open persistence
 		try {
 			message.setMessageId(getNextMessageId());
+			key = getSendBufferedPersistenceKey(message);
 			try {
 				persistence.put(key, (MqttPublish) message);
 			} catch (MqttPersistenceException mpe){
@@ -1000,7 +1002,6 @@ protected void notifyReceivedAck(MqttAck ack) throws MqttException {
 				queueLock.notifyAll();
 			}
 		} else {
-			// Sub ack or unsuback
 			notifyResult(ack, token, mex);
 			releaseMessageId(ack.getMessageId());
 			tokenStore.removeToken(ack);

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

@@ -122,5 +122,9 @@ public void run() {
 	public void setPublishCallback(IDisconnectedBufferCallback callback) {
 		this.callback = callback;
 	}
+	
+	public boolean isPersistBuffer(){
+		return bufferOpts.isPersistBuffer();
+	}
 
 }