Automatic Reconnect & Offline Buffering functionality

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

Author: jpwsutton

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

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2009, 2015 IBM Corp.
+ * Copyright (c) 2009, 2016 IBM Corp.
  *
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
@@ -15,21 +15,25 @@
  *    Ian Craggs - per subscription message handlers (bug 466579)
  *    Ian Craggs - ack control (bug 472172)
  *    James Sutton - checkForActivity Token (bug 473928)
+ *    James Sutton - Automatic Reconnect & Offline Buffering.
  */
 package org.eclipse.paho.client.mqttv3.internal;
 
 import java.util.Enumeration;
 import java.util.Properties;
 import java.util.Vector;
 
+import org.eclipse.paho.client.mqttv3.BufferedMessage;
 import org.eclipse.paho.client.mqttv3.IMqttActionListener;
 import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
 import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
 import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
 import org.eclipse.paho.client.mqttv3.MqttPingSender;
 import org.eclipse.paho.client.mqttv3.MqttToken;
@@ -74,7 +78,9 @@ public class ClientComms {
 	private byte	conState = DISCONNECTED;
 	private Object	conLock = new Object();  	// Used to synchronize connection state
 	private boolean	closePending = false;
-
+	private boolean resting = false;
+	private DisconnectedMessageBuffer disconnectedMessageBuffer;
+	
 	/**
 	 * Creates a new ClientComms object, using the specified module to handle
 	 * the network calls.
@@ -141,7 +147,19 @@ public void sendNoWait(MqttWireMessage message, MqttToken token) throws MqttExce
 		if (isConnected() ||
 				(!isConnected() && message instanceof MqttConnect) ||
 				(isDisconnecting() && message instanceof MqttDisconnect)) {
-			this.internalSend(message, token);
+			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);
+				disconnectedMessageBuffer.putMessage(message, token);
+			} else {
+				this.internalSend(message, token);
+			}
+		} else if(disconnectedMessageBuffer != null && isResting()){
+			//@TRACE 508=Client Resting, Offline Buffer available. Adding message to buffer. message={0}
+			log.fine(CLASS_NAME, methodName, "508", new Object[] {message.getKey()});
+			this.clientState.persistBufferedMessage(message);
+			disconnectedMessageBuffer.putMessage(message, token);
 		} else {
 			//@TRACE 208=failed: not connected
 			log.fine(CLASS_NAME, methodName, "208");
@@ -336,7 +354,10 @@ public void shutdownConnection(MqttToken token, MqttException reason) {
 		}
 
 		try {
-			if (persistence != null) {persistence.close();}
+			if(disconnectedMessageBuffer == null && persistence != null){
+				persistence.close();
+			}
+			
 		}catch(Exception ex) {
 			// Ignore as we are shutting down
 		}
@@ -499,12 +520,22 @@ public boolean isClosed() {
 			return conState == CLOSED;
 		}
 	}
+	
+	public boolean isResting() {
+		synchronized (conLock) {
+			return resting;
+		}
+	}
 
 
 	public void setCallback(MqttCallback mqttCallback) {
 		this.callback.setCallback(mqttCallback);
 	}
 
+	public void setReconnectCallback(MqttCallbackExtended callback){
+		this.callback.setReconnectCallback(callback);
+	}
+	
 	public void setManualAcks(boolean manualAcks) {
 		this.callback.setManualAcks(manualAcks);
 	}
@@ -720,4 +751,62 @@ private void handleRunException(Exception ex) {
 
 		shutdownConnection(null, mex);
 	}
+
+	/**
+	 * When Automatic reconnect is enabled, we want ClientComs to enter the
+	 * 'resting' state if disconnected. This will allow us to publish messages
+	 * @param resting
+	 */
+	public void setRestingState(boolean resting) {
+		this.resting = resting;
+	}
+
+	public void setDisconnectedMessageBuffer(DisconnectedMessageBuffer disconnectedMessageBuffer) {
+		this.disconnectedMessageBuffer = disconnectedMessageBuffer;
+	}
+	
+	public int getBufferedMessageCount(){
+		return this.disconnectedMessageBuffer.getMessageCount();
+	}
+	
+	public MqttMessage getBufferedMessage(int bufferIndex){
+		MqttPublish send = (MqttPublish) this.disconnectedMessageBuffer.getMessage(bufferIndex).getMessage();
+		return send.getMessage();
+	}
+	
+	public void deleteBufferedMessage(int bufferIndex){
+		this.disconnectedMessageBuffer.deleteMessage(bufferIndex);
+	}
+	
+
+	/**
+	 * When the client automatically reconnects, we want to send all messages from the
+	 * buffer first before allowing the user to send any messages
+	 * @throws MqttException 
+	 */
+	public void notifyReconnect() {
+		final String methodName = "notifyReconnect";
+		if(disconnectedMessageBuffer != null){
+			//@TRACE 509=Client Reconnected, Offline Buffer Available. Sending Buffered Messages.
+			log.fine(CLASS_NAME, methodName, "509");
+			disconnectedMessageBuffer.setPublishCallback(new IDisconnectedBufferCallback() {
+				
+				public void publishBufferedMessage(BufferedMessage bufferedMessage) throws MqttException {
+					if (isConnected()) {
+						//@TRACE 510=Publising 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 {
+						//@TRACE 208=failed: not connected
+						log.fine(CLASS_NAME, methodName, "208");
+						throw ExceptionHelper.createMqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
+					}	
+				}
+			});
+			new Thread(disconnectedMessageBuffer).start();
+		}
+	}
+
 }

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

@@ -0,0 +1,42 @@
+/*******************************************************************************
+ * Copyright (c) 2016 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution. 
+ *
+ * The Eclipse Public License is available at 
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at 
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    James Sutton - Initial Contribution for Automatic Reconnect & Offline Buffering
+ */
+package org.eclipse.paho.client.mqttv3;
+
+import org.eclipse.paho.client.mqttv3.internal.wire.MqttWireMessage;
+
+/**
+ * A BufferedMessage contains an MqttWire Message and token
+ * it allows both message and token to be buffered when the client
+ * is in resting state
+ */
+public class BufferedMessage {
+	
+	private MqttWireMessage message;
+	private MqttToken token; 
+	
+	public BufferedMessage(MqttWireMessage message, MqttToken token){
+		this.message = message;
+		this.token = token;
+	}
+
+	public MqttWireMessage getMessage() {
+		return message;
+	}
+
+	public MqttToken getToken() {
+		return token;
+	}
+}

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

@@ -0,0 +1,90 @@
+/*******************************************************************************
+ * Copyright (c) 2016 IBM Corp.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution. 
+ *
+ * The Eclipse Public License is available at 
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at 
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * Contributors:
+ *    James Sutton - Initial Contribution for Automatic Reconnect & Offline Buffering
+ */
+package org.eclipse.paho.client.mqttv3;
+
+/**
+ * Holds the set of options that govern the behaviour
+ * of Offline (or Disconnected) buffering of messages
+ */
+public class DisconnectedBufferOptions {
+	
+	/**
+	 * The default size of the disconnected buffer
+	 */
+	public static final int DISCONNECTED_BUFFER_SIZE_DEFAULT = 5000;
+	
+	public static final boolean DISCONNECTED_BUFFER_ENABLED_DEFAULT = false;
+	
+	public static final boolean PERSIST_DISCONNECTED_BUFFER_DEFAULT = false;
+	
+	public static final boolean DELETE_OLDEST_MESSAGES_DEFAULT = false;
+	
+	private int bufferSize = DISCONNECTED_BUFFER_SIZE_DEFAULT;
+	private boolean bufferEnabled = DISCONNECTED_BUFFER_ENABLED_DEFAULT;
+	private boolean persistBuffer = PERSIST_DISCONNECTED_BUFFER_DEFAULT;
+	private boolean deleteOldestMessages = DELETE_OLDEST_MESSAGES_DEFAULT;
+	
+	/**
+	 * Constructs a new <code>DisconnectedBufferOptions</code> object using the
+	 * default values.
+	 *
+	 * The defaults are:
+	 * <ul>
+	 * <li>The disconnected buffer is disabled</li>
+	 * <li>The buffer holds 5000 messages</li>
+	 * <li>The buffer is not persisted</li>
+	 * <li>Once the buffer is full, old messages are not deleted</li>
+	 * </ul>
+	 * More information about these values can be found in the setter methods.
+	 */
+	public DisconnectedBufferOptions() {
+	}
+
+	public int getBufferSize() {
+		return bufferSize;
+	}
+
+	public void setBufferSize(int bufferSize) {
+		if (bufferSize < 1) {
+            throw new IllegalArgumentException();
+        }
+		this.bufferSize = bufferSize;
+	}
+
+	public boolean isBufferEnabled() {
+		return bufferEnabled;
+	}
+
+	public void setBufferEnabled(boolean bufferEnabled) {
+		this.bufferEnabled = bufferEnabled;
+	}
+
+	public boolean isPersistBuffer() {
+		return persistBuffer;
+	}
+
+	public void setPersistBuffer(boolean persistBuffer) {
+		this.persistBuffer = persistBuffer;
+	}
+
+	public boolean isDeleteOldestMessages() {
+		return deleteOldestMessages;
+	}
+
+	public void setDeleteOldestMessages(boolean deleteOldestMessages) {
+		this.deleteOldestMessages = deleteOldestMessages;
+	}
+}