Merge pull request #760 from rdasgupt/fix596rework

Fix conflicts in PR 596

Author: Ranjan Dasgupta

org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/automaticReconnect/OfflineBufferingTest.java

@@ -29,6 +29,7 @@
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
 import org.eclipse.paho.client.mqttv3.internal.wire.MqttPublish;
 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 import org.eclipse.paho.client.mqttv3.test.logging.LoggingUtilities;
@@ -312,6 +313,75 @@ public void testDeleteOldestBufferedMessages() throws Exception {
 		proxy.disableProxy();
 	}
 
+
+        /**
+	 * Tests that the buffer correctly handles persisted messages being buffered when the
+	 * buffer is full and deleteOldestBufferedMessage is set to true.
+	 */
+	@Test
+	public void testDeleteOldestBufferedMessagesWithPersistance() throws Exception {
+		String methodName = Utility.getMethodName();
+		LoggingUtilities.banner(log, cclass, methodName);
+		int maxMessages = 10;
+
+		// Tokens
+		IMqttToken connectToken;
+
+		MqttClientPersistence persistence = new MemoryPersistence();
+
+		// Client Options
+		MqttConnectOptions options = new MqttConnectOptions();
+		options.setCleanSession(false);
+		options.setAutomaticReconnect(false);
+		MqttAsyncClient client = new MqttAsyncClient("tcp://localhost:" + proxy.getLocalPort(), methodName, persistence);
+		DisconnectedBufferOptions disconnectedOpts = new DisconnectedBufferOptions();
+		disconnectedOpts.setBufferEnabled(true);
+		// Set buffer to 100 to save time
+		disconnectedOpts.setBufferSize(maxMessages);
+		disconnectedOpts.setPersistBuffer(true);
+		disconnectedOpts.setDeleteOldestMessages(true);
+		client.setBufferOpts(disconnectedOpts);
+
+		// Enable Proxy & Connect to server
+		proxy.enableProxy();
+		connectToken = client.connect(options);
+		connectToken.waitForCompletion();
+		boolean isConnected = client.isConnected();
+		log.info("First Connection isConnected: " + isConnected);
+		Assert.assertTrue(isConnected);
+
+		// Disable Proxy and cause disconnect
+		proxy.disableProxy();
+		isConnected = client.isConnected();
+		log.info("Proxy Disconnect isConnected: " + isConnected);
+		Assert.assertFalse(isConnected);
+
+		int x;
+
+		// Publish 10 messages
+		for (x = 0; x < maxMessages; x++) {
+			MqttMessage message = new MqttMessage(Integer.toString(x).getBytes());
+			client.publish(topicPrefix + methodName, message);
+		}
+
+		// Publish one message too many
+		log.info("About to publish one message too many");
+		client.publish(topicPrefix + methodName, new MqttMessage(Integer.toString(x).getBytes()));
+
+		Assert.assertFalse(persistence.containsKey("sb-1"));
+		Assert.assertTrue(persistence.containsKey("sb-2"));
+		Assert.assertTrue(persistence.containsKey("sb-11"));
+
+		// Make sure that the message now at index 0 in the buffer is '1'
+		// instead of '0'
+		MqttMessage messageAt0 = client.getBufferedMessage(0);
+		String msg = new String(messageAt0.getPayload());
+		Assert.assertEquals("1", msg);
+		client.close();
+		client = null;
+		proxy.disableProxy();
+	}
+
 	/**
 	 * Tests that A message cannot be buffered when the buffer is full and
 	 * deleteOldestBufferedMessage is set to false.

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

@@ -874,13 +874,26 @@ public void notifyConnect() {
 			log.fine(CLASS_NAME, methodName, "509", null);
 
 			disconnectedMessageBuffer.setPublishCallback(new ReconnectDisconnectedBufferCallback(methodName));
+                        disconnectedMessageBuffer.setMessageDiscardedCallBack(new MessageDiscardedCallback());
 			if (executorService == null) {
 				new Thread(disconnectedMessageBuffer).start();
 			} else {
 				executorService.execute(disconnectedMessageBuffer);
 			}
 		}
 	}
+
+
+	class MessageDiscardedCallback implements IDiscardedBufferMessageCallback {
+
+		@Override
+		public void messageDiscarded(MqttWireMessage message) {
+			if(disconnectedMessageBuffer.isPersistBuffer()) {
+				clientState.unPersistBufferedMessage(message);
+			}
+		}
+	}
+
 
 	class ReconnectDisconnectedBufferCallback implements IDisconnectedBufferCallback{
 

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

@@ -33,6 +33,7 @@ public class DisconnectedMessageBuffer implements Runnable {
 	private ArrayList<BufferedMessage> buffer;
 	private final Object bufLock = new Object(); // Used to synchronise the buffer
 	private IDisconnectedBufferCallback callback;
+        private IDiscardedBufferMessageCallback messageDiscardedCallBack;
 
 	public DisconnectedMessageBuffer(DisconnectedBufferOptions options) {
 		this.bufferOpts = options;
@@ -63,6 +64,10 @@ public void putMessage(MqttWireMessage message, MqttToken token) throws MqttExce
 			if (buffer.size() < bufferOpts.getBufferSize()) {
 				buffer.add(bufferedMessage);
 			} else if (bufferOpts.isDeleteOldestMessages() == true) {
+				if(messageDiscardedCallBack != null){
+					BufferedMessage discardedMessage = (BufferedMessage) buffer.get(0);
+					messageDiscardedCallBack.messageDiscarded(discardedMessage.getMessage());
+				}
 				buffer.remove(0);
 				buffer.add(bufferedMessage);
 			} else {
@@ -146,4 +151,8 @@ public boolean isPersistBuffer() {
 		return bufferOpts.isPersistBuffer();
 	}
 
+        public void setMessageDiscardedCallBack(IDiscardedBufferMessageCallback callback) {
+		this.messageDiscardedCallBack = callback;
+	}
+
 }

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

@@ -0,0 +1,23 @@
+/*******************************************************************************
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v2.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    https://www.eclipse.org/legal/epl-2.0
+ * and the Eclipse Distribution License is available at
+ *   https://www.eclipse.org/org/documents/edl-v10.php
+ *
+ * Contributors:
+ *    alexm   26/10/18
+ */
+
+package org.eclipse.paho.client.mqttv3.internal;
+
+import org.eclipse.paho.client.mqttv3.internal.wire.MqttWireMessage;
+
+public interface IDiscardedBufferMessageCallback {
+    void messageDiscarded(MqttWireMessage message);
+}
+