Issue #553 - MQTT-3.1.2-21 - Using server keep alive if provided.

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

Author: jpwsutton

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/ClientComms.java

@@ -340,8 +340,7 @@ public void connect(MqttConnectionOptions options, MqttToken token) throws MqttE
 				 * conOptions.getUserName(), conOptions.getPassword(),
 				 * conOptions.getWillMessage(), conOptions.getWillDestination()
 				 */
-
-				this.clientState.setKeepAliveSecs(conOptions.getKeepAliveInterval());
+				this.mqttConnection.setKeepAliveSeconds(conOptions.getKeepAliveInterval());
 				this.clientState.setCleanStart(conOptions.isCleanStart());
 
 				tokenStore.open();
@@ -728,7 +727,7 @@ public MqttClientInterface getClient() {
 	}
 
 	public long getKeepAlive() {
-		return this.clientState.getKeepAlive();
+		return this.mqttConnection.getKeepAlive();
 	}
 
 	public MqttState getClientState() {

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

@@ -118,7 +118,7 @@ public class ClientState implements MqttState {
 	private CommsTokenStore tokenStore;
 	private ClientComms clientComms = null;
 	private CommsCallback callback = null;
-	private long keepAlive;
+	//private long keepAlive;
 	private boolean cleanStart;
 	private MqttClientPersistence persistence;
 
@@ -181,14 +181,6 @@ protected ClientState(MqttClientPersistence persistence, CommsTokenStore tokenSt
 		restoreState();
 	}
 
-	protected void setKeepAliveSecs(long keepAliveSecs) {
-		this.keepAlive = keepAliveSecs * 1000;
-	}
-
-	protected long getKeepAlive() {
-		return this.keepAlive;
-	}
-
 	protected void setCleanStart(boolean cleanStart) {
 		this.cleanStart = cleanStart;
 	}
@@ -712,9 +704,10 @@ public MqttToken checkForActivity(MqttActionListener pingCallback) throws MqttEx
 		}
 
 		MqttToken token = null;
-		long nextPingTime = getKeepAlive();
+		long nextPingTime = this.mqttConnection.getKeepAlive();
+		long keepAlive = this.mqttConnection.getKeepAlive();
 
-		if (connected && this.keepAlive > 0) {
+		if (connected && this.mqttConnection.getKeepAlive() > 0) {
 			long time = System.currentTimeMillis();
 			// Reduce schedule frequency since System.currentTimeMillis is no accurate, add
 			// a buffer
@@ -733,7 +726,7 @@ public MqttToken checkForActivity(MqttActionListener pingCallback) throws MqttEx
 					// @TRACE 619=Timed out as no activity, keepAlive={0} lastOutboundActivity={1}
 					// lastInboundActivity={2} time={3} lastPing={4}
 					log.severe(CLASS_NAME, methodName, "619",
-							new Object[] { Long.valueOf(this.keepAlive), Long.valueOf(lastOutboundActivity),
+							new Object[] { Long.valueOf(keepAlive), Long.valueOf(lastOutboundActivity),
 									Long.valueOf(lastInboundActivity), Long.valueOf(time), Long.valueOf(lastPing) });
 
 					// A ping has already been sent. At this point, assume that the
@@ -748,7 +741,7 @@ public MqttToken checkForActivity(MqttActionListener pingCallback) throws MqttEx
 					// I am probably blocked on a write operations as I should have been able to
 					// write at least a ping message
 					log.severe(CLASS_NAME, methodName, "642",
-							new Object[] { Long.valueOf(this.keepAlive), Long.valueOf(lastOutboundActivity),
+							new Object[] { Long.valueOf(keepAlive), Long.valueOf(lastOutboundActivity),
 									Long.valueOf(lastInboundActivity), Long.valueOf(time), Long.valueOf(lastPing) });
 
 					// A ping has not been sent but I am not progressing on the current write
@@ -776,7 +769,7 @@ public MqttToken checkForActivity(MqttActionListener pingCallback) throws MqttEx
 
 					// @TRACE 620=ping needed. keepAlive={0} lastOutboundActivity={1}
 					// lastInboundActivity={2}
-					log.fine(CLASS_NAME, methodName, "620", new Object[] { Long.valueOf(this.keepAlive),
+					log.fine(CLASS_NAME, methodName, "620", new Object[] { Long.valueOf(keepAlive),
 							Long.valueOf(lastOutboundActivity), Long.valueOf(lastInboundActivity) });
 
 					// pingOutstanding++; // it will be set after the ping has been written on the
@@ -790,13 +783,13 @@ public MqttToken checkForActivity(MqttActionListener pingCallback) throws MqttEx
 					tokenStore.saveToken(token, pingCommand);
 					pendingFlows.insertElementAt(pingCommand, 0);
 
-					nextPingTime = getKeepAlive();
+					nextPingTime = this.mqttConnection.getKeepAlive();
 
 					// Wake sender thread since it may be in wait state (in ClientState.get())
 					notifyQueueLock();
 				} else {
 					log.fine(CLASS_NAME, methodName, "634", null);
-					nextPingTime = Math.max(1, getKeepAlive() - (time - lastOutboundActivity));
+					nextPingTime = Math.max(1, this.mqttConnection.getKeepAlive() - (time - lastOutboundActivity));
 				}
 			}
 			// @TRACE 624=Schedule next ping at {0}
@@ -895,17 +888,6 @@ protected MqttWireMessage get() throws MqttException {
 		return result;
 	}
 
-	/*
-	 * (non-Javadoc)
-	 * 
-	 * @see
-	 * org.eclipse.paho.mqttv5.client.internal.MqttState#setKeepAliveInterval(long)
-	 */
-	@Override
-	public void setKeepAliveInterval(long interval) {
-		this.keepAlive = interval;
-	}
-
 	/*
 	 * (non-Javadoc)
 	 * 

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/ConnectActionListener.java

@@ -124,6 +124,11 @@ public void onSuccess(IMqttToken token) {
 		mqttConnection.setSubscriptionIdentifiersAvailable(
 				myToken.getMessageProperties().isSubscriptionIdentifiersAvailable());
 		mqttConnection.setSharedSubscriptionsAvailable(myToken.getMessageProperties().isSharedSubscriptionAvailable());
+		
+		// If provided, set the server keep alive value.
+		if(myToken.getMessageProperties().getServerKeepAlive() != null) {
+			mqttConnection.setKeepAliveSeconds(myToken.getMessageProperties().getServerKeepAlive());
+		}
 
 		// If we are assigning the client ID post connect, then we need to re-initialise
 		// our persistence layer.

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/MqttConnectionState.java

@@ -36,6 +36,7 @@ public class MqttConnectionState {
 	private Boolean subscriptionIdentifiersAvailable = true;
 	private Boolean sharedSubscriptionsAvailable = true;
 	private boolean sendReasonMessages = false;
+	private long keepAlive = 60;
 
 	// ******* Counters ******//
 	private AtomicInteger nextOutgoingTopicAlias = new AtomicInteger(1);
@@ -138,4 +139,14 @@ public void setSendReasonMessages(boolean enableReasonMessages) {
 		this.sendReasonMessages = enableReasonMessages;
 	}
 
+
+	public long getKeepAlive() {
+		return keepAlive;
+	}
+
+
+	public void setKeepAliveSeconds(long keepAlive) {
+		this.keepAlive = keepAlive * 1000;
+	}
+
 }

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/MqttState.java

@@ -66,8 +66,6 @@ public interface MqttState {
 	 */
 	MqttToken checkForActivity(MqttActionListener pingCallback) throws MqttException;
 
-	void setKeepAliveInterval(long interval);
-
 	void notifySentBytes(int sentBytesCount);
 
 	void notifyReceivedBytes(int receivedBytesCount);