Issue #582 - Tweaking Offline Buffer test to work, fixing trace in multiple locations

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

Author: jpwsutton

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

@@ -63,7 +63,7 @@ public static void setUpBeforeClass() throws Exception {
 			LoggingUtilities.banner(log, cclass, methodName);
 			serverURI = TestProperties.getServerURI();
 			serverURIString = "tcp://" + serverURI.getHost() + ":" + serverURI.getPort();
-		    topicPrefix = "OfflineBufferingTest-" + UUID.randomUUID().toString() + "-";
+			topicPrefix = "OfflineBufferingTest-" + UUID.randomUUID().toString() + "-";
 
 			// Use 0 for the first time.
 			proxy = new ConnectionManipulationProxyServer(serverURI.getHost(), serverURI.getPort(), 2883);
@@ -78,16 +78,15 @@ public static void setUpBeforeClass() throws Exception {
 		}
 
 	}
-	
+
 	@After
 	public void clearUpAfterTest() {
 		proxy.disableProxy();
 	}
 
 	/**
-	 * Tests that A message can be buffered whilst the client is in a
-	 * disconnected state and is then delivered once the client has reconnected
-	 * automatically.
+	 * Tests that A message can be buffered whilst the client is in a disconnected
+	 * state and is then delivered once the client has reconnected automatically.
 	 */
 	@Test
 	public void testSingleMessageBufferAndDeliver() throws Exception {
@@ -157,8 +156,8 @@ public void testSingleMessageBufferAndDeliver() throws Exception {
 
 	/**
 	 * Tests that multiple messages can be buffered whilst the client is in a
-	 * disconnected state and that they are all then delivered once the client
-	 * has connected automatically.
+	 * disconnected state and that they are all then delivered once the client has
+	 * connected automatically.
 	 */
 	@Test
 	public void testManyMessageBufferAndDeliver() throws Exception {
@@ -172,13 +171,18 @@ public void testManyMessageBufferAndDeliver() throws Exception {
 		MqttConnectOptions options = new MqttConnectOptions();
 		options.setCleanSession(true);
 		options.setAutomaticReconnect(true);
+		
+		// Workaround for Issue #582 - Remove once fixed.
+		options.setMaxInflight(100);
+		
 		MqttAsyncClient client = new MqttAsyncClient("tcp://localhost:" + proxy.getLocalPort(), methodName, DATA_STORE);
 		DisconnectedBufferOptions disconnectedOpts = new DisconnectedBufferOptions();
 		disconnectedOpts.setBufferEnabled(true);
 		client.setBufferOpts(disconnectedOpts);
 
 		// Create subscription client that won't be affected by proxy
-		MqttAsyncClient subClient = new MqttAsyncClient(serverURIString, methodName + "sub-client");
+		MemoryPersistence persistence = new MemoryPersistence();
+		MqttAsyncClient subClient = new MqttAsyncClient(serverURIString, methodName + "sub-client", persistence);
 		MqttV3Receiver mqttV3Receiver = new MqttV3Receiver(subClient, LoggingUtilities.getPrintStream());
 		subClient.setCallback(mqttV3Receiver);
 		IMqttToken subConnectToken = subClient.connect();
@@ -226,12 +230,13 @@ public void testManyMessageBufferAndDeliver() throws Exception {
 		isConnected = client.isConnected();
 		log.info("Proxy Re-Enabled isConnected: " + isConnected);
 		Assert.assertTrue(isConnected);
-		
+
 		Thread.sleep(5000);
 
 		// Check that all messages have been delivered
 		for (int x = 0; x < 100; x++) {
-			boolean recieved = mqttV3Receiver.validateReceipt(topicPrefix + methodName, 1, Integer.toString(x).getBytes());
+			boolean recieved = mqttV3Receiver.validateReceipt(topicPrefix + methodName, 1,
+					Integer.toString(x).getBytes());
 			Assert.assertTrue(recieved);
 		}
 		log.info("All messages sent and Recieved correctly.");
@@ -447,7 +452,7 @@ public void testUnPersistBufferedMessagesOnNewClient() throws Exception {
 		List<String> persistedKeys = Collections.list(persistence.keys());
 		log.info("There are now: " + persistedKeys.size() + " keys in persistence");
 		Assert.assertEquals(1, persistedKeys.size());
-	
+
 		// Create Subscription client to watch for the message being published
 		// as soon as the main client connects
 		log.info("Creating subscription client");

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

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2016 IBM Corp.
+ * Copyright (c) 2009, 2018 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -56,8 +56,8 @@
 public class ClientComms {
 	public static String 		VERSION = "${project.version}";
 	public static String 		BUILD_LEVEL = "L${build.level}";
-	private static final String CLASS_NAME = ClientComms.class.getName();
-	private static final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT,CLASS_NAME);
+	private final String CLASS_NAME = ClientComms.class.getName();
+	private final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT,CLASS_NAME);
 
 	private static final byte CONNECTED	= 0;
 	private static final byte CONNECTING	= 1;
@@ -854,7 +854,7 @@ public void notifyConnect() {
 		final String methodName = "notifyConnect";
 		if(disconnectedMessageBuffer != null){
 			//@TRACE 509=Client Connected, Offline Buffer Available. Sending Buffered Messages.
-			log.fine(CLASS_NAME, methodName, "509");
+			log.fine(CLASS_NAME, methodName, "509", null);
 
 			disconnectedMessageBuffer.setPublishCallback(new ReconnectDisconnectedBufferCallback(methodName));
 			executorService.execute(disconnectedMessageBuffer);
@@ -871,14 +871,16 @@ class ReconnectDisconnectedBufferCallback implements IDisconnectedBufferCallback
 
 		public void publishBufferedMessage(BufferedMessage bufferedMessage) throws MqttException {
 			if (isConnected()) {
-				while(clientState.getActualInFlight() >= (clientState.getMaxInFlight()-1)){
+				// First pass at making sure that we don't flood the in-flight messages
+				while(clientState.getActualInFlight() >= (clientState.getMaxInFlight()-3)){
 					// We need to Yield to the other threads to allow the in flight messages to clear
 					Thread.yield();
 
 				}
-				//@TRACE 510=Publising Buffered message message={0}
+				//@TRACE 510=Publishing Buffered message message={0}
 				log.fine(CLASS_NAME, methodName, "510", new Object[] {bufferedMessage.getMessage().getKey()});
 				internalSend(bufferedMessage.getMessage(), bufferedMessage.getToken());
+				
 				// Delete from persistence if in there
 				clientState.unPersistBufferedMessage(bufferedMessage.getMessage());
 			} else {

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

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2015 IBM Corp.
+ * Copyright (c) 2009, 2018 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -58,9 +58,7 @@
  *
  * @see IMqttClient
  */
-public class MqttClient implements IMqttClient { //), DestinationProvider {
-	//private static final String CLASS_NAME = MqttClient.class.getName();
-	//private static final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT,CLASS_NAME);
+public class MqttClient implements IMqttClient { 
 
 	protected MqttAsyncClient aClient = null;  // Delegate implementation to MqttAsyncClient
 	protected long timeToWait = -1;				// How long each method should wait for action to complete

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

@@ -33,7 +33,7 @@
  */
 public class ScheduledExecutorPingSender implements MqttPingSender {
 	private static final String CLASS_NAME = ScheduledExecutorPingSender.class.getName();
-	private static final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
+	private final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
 
 	private ClientComms comms;
 	private ScheduledExecutorService executorService;

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

@@ -720,7 +720,7 @@ public MqttToken checkForActivity(IMqttActionListener pingCallback) throws MqttE
 			// Below might not be necessary since move to nanoTime (Issue #278)
 			//Reduce schedule frequency since System.currentTimeMillis is no accurate, add a buffer
 			//It is 1/10 in minimum keepalive unit.
-			int delta = 1000;
+			int delta = 100000;
 
 			// ref bug: https://bugs.eclipse.org/bugs/show_bug.cgi?id=446663
             synchronized (pingOutstandingLock) {
@@ -774,14 +774,15 @@ public MqttToken checkForActivity(IMqttActionListener pingCallback) throws MqttE
                     tokenStore.saveToken(token, pingCommand);
                     pendingFlows.insertElementAt(pingCommand, 0);
 
-                    nextPingTime = getKeepAlive();
+                    nextPingTime = this.keepAlive;
 
                     //Wake sender thread since it may be in wait state (in ClientState.get())                                                                                                                             
                     notifyQueueLock();
                 }
                 else {
+                		//@TRACE 634=ping not needed yet. Schedule next ping.
                     log.fine(CLASS_NAME, methodName, "634", null);
-                    nextPingTime = Math.max(1, getKeepAlive() - (time - lastOutboundActivity));
+                    nextPingTime = Math.max(1, this.keepAlive - (time - lastOutboundActivity));
                 }
             }
             //@TRACE 624=Schedule next ping at {0}                                                                                                                                                                                

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

@@ -47,7 +47,7 @@
  */
 public class CommsCallback implements Runnable {
 	private static final String CLASS_NAME = CommsCallback.class.getName();
-	private static final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
+	private final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
 
 	private static final int INBOUND_QUEUE_SIZE = 10;
 	private MqttCallback mqttCallback;

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

@@ -26,102 +26,112 @@
 import org.eclipse.paho.client.mqttv3.logging.LoggerFactory;
 
 public class DisconnectedMessageBuffer implements Runnable {
-	
-	private static final String CLASS_NAME = "DisconnectedMessageBuffer";
+
+	private final String CLASS_NAME = "DisconnectedMessageBuffer";
 	private Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
 	private DisconnectedBufferOptions bufferOpts;
 	private ArrayList<BufferedMessage> buffer;
-	private Object	bufLock = new Object();  	// Used to synchronise the buffer
+	private Object bufLock = new Object(); // Used to synchronise the buffer
 	private IDisconnectedBufferCallback callback;
-	
-	public DisconnectedMessageBuffer(DisconnectedBufferOptions options){
+
+	public DisconnectedMessageBuffer(DisconnectedBufferOptions options) {
 		this.bufferOpts = options;
 		buffer = new ArrayList<BufferedMessage>();
 	}
-	
+
 	/**
-	 * This will add a new message to the offline buffer,
-	 * if the buffer is full and deleteOldestMessages is enabled
-	 * then the 0th item in the buffer will be deleted and the
-	 * new message will be added. If it is not enabled then an
+	 * This will add a new message to the offline buffer, if the buffer is full and
+	 * deleteOldestMessages is enabled then the 0th item in the buffer will be
+	 * deleted and the new message will be added. If it is not enabled then an
 	 * MqttException will be thrown.
-	 * @param message the {@link MqttWireMessage} that will be buffered
-	 * @param token the associated {@link MqttToken}
-	 * @throws MqttException if the Buffer is full
+	 * 
+	 * @param message
+	 *            the {@link MqttWireMessage} that will be buffered
+	 * @param token
+	 *            the associated {@link MqttToken}
+	 * @throws MqttException
+	 *             if the Buffer is full
 	 */
-	public void putMessage(MqttWireMessage message, MqttToken token) throws MqttException{
+	public void putMessage(MqttWireMessage message, MqttToken token) throws MqttException {
 		BufferedMessage bufferedMessage = new BufferedMessage(message, token);
 		synchronized (bufLock) {
-			if(buffer.size() < bufferOpts.getBufferSize()){
+			if (buffer.size() < bufferOpts.getBufferSize()) {
 				buffer.add(bufferedMessage);
-			} else if(bufferOpts.isDeleteOldestMessages() == true){
+			} else if (bufferOpts.isDeleteOldestMessages() == true) {
 				buffer.remove(0);
 				buffer.add(bufferedMessage);
-			}else {
+			} else {
 				throw new MqttException(MqttException.REASON_CODE_DISCONNECTED_BUFFER_FULL);
 			}
 		}
 	}
-	
+
 	/**
 	 * Retrieves a message from the buffer at the given index.
-	 * @param messageIndex the index of the message to be retrieved in the buffer
+	 * 
+	 * @param messageIndex
+	 *            the index of the message to be retrieved in the buffer
 	 * @return the {@link BufferedMessage}
 	 */
-	public BufferedMessage getMessage(int messageIndex){
+	public BufferedMessage getMessage(int messageIndex) {
 		synchronized (bufLock) {
-			return((BufferedMessage) buffer.get(messageIndex));
+			return ((BufferedMessage) buffer.get(messageIndex));
 		}
 	}
-	
-	
+
 	/**
 	 * Removes a message from the buffer
-	 * @param messageIndex the index of the message to be deleted in the buffer
+	 * 
+	 * @param messageIndex
+	 *            the index of the message to be deleted in the buffer
 	 */
-	public void deleteMessage(int messageIndex){
+	public void deleteMessage(int messageIndex) {
 		synchronized (bufLock) {
 			buffer.remove(messageIndex);
 		}
 	}
-	
+
 	/**
 	 * Returns the number of messages currently in the buffer
+	 * 
 	 * @return The count of messages in the buffer
 	 */
 	public int getMessageCount() {
 		synchronized (bufLock) {
 			return buffer.size();
 		}
 	}
-	
+
 	/**
 	 * Flushes the buffer of messages into an open connection
 	 */
 	public void run() {
 		final String methodName = "run";
 		// @TRACE 516=Restoring all buffered messages.
 		log.fine(CLASS_NAME, methodName, "516");
-			while(getMessageCount() > 0){
-				try {
+		while (getMessageCount() > 0) {
+			try {
 				BufferedMessage bufferedMessage = getMessage(0);
 				callback.publishBufferedMessage(bufferedMessage);
 				// Publish was successful, remove message from buffer.
 				deleteMessage(0);
-				} catch (MqttException ex) {
-					// Error occurred attempting to publish buffered message likely because the client is not connected
-					// @TRACE 519=Error occurred attempting to publish buffered message due to disconnect. Exception: {0}.
-					log.warning(CLASS_NAME, methodName, "519", new Object[]{ex.getMessage()});
-					break;
-				}
+			} catch (MqttException ex) {
+				// Error occurred attempting to publish buffered message likely because the
+				// client is not connected
+				// @TRACE 519=Error occurred attempting to publish buffered message due to disconnect. Exception: {0}:{1}.
+				log.severe(CLASS_NAME, methodName, "519", new Object[] { ex.getReasonCode(), ex.getMessage() });
+				break;
+
 			}
+
+		}
 	}
 
 	public void setPublishCallback(IDisconnectedBufferCallback callback) {
 		this.callback = callback;
 	}
-	
-	public boolean isPersistBuffer(){
+
+	public boolean isPersistBuffer() {
 		return bufferOpts.isPersistBuffer();
 	}
 

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

@@ -34,8 +34,8 @@
  * <code>MqttWireMessage</code>. 
  */
 public class MqttInputStream extends InputStream {
-	private static final String CLASS_NAME = MqttInputStream.class.getName();
-	private static final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
+	private final String CLASS_NAME = MqttInputStream.class.getName();
+	private final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
 
 	private ClientState clientState = null;
 	private DataInputStream in;	
@@ -117,8 +117,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 301= received {0} 
+				log.fine(CLASS_NAME, methodName, "301",new Object[] {message});
 			}
 		} catch (SocketTimeoutException e) {
 			// ignore socket read timeout