Merge pull request #775 from 51systems/abstract-hrt

Implementation specific timer for ClientState

Author: Ranjan Dasgupta

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

@@ -95,7 +95,8 @@ public class ClientComms {
 	 * @param executorService the {@link ExecutorService}
 	 * @throws MqttException if an exception occurs whilst communicating with the server
 	 */
-	public ClientComms(IMqttAsyncClient client, MqttClientPersistence persistence, MqttPingSender pingSender, ExecutorService executorService) throws MqttException {
+	public ClientComms(IMqttAsyncClient client, MqttClientPersistence persistence, MqttPingSender pingSender,
+					   ExecutorService executorService, HighResolutionTimer highResolutionTimer) throws MqttException {
 		this.conState = DISCONNECTED;
 		this.client 	= client;
 		this.persistence = persistence;
@@ -105,7 +106,7 @@ public ClientComms(IMqttAsyncClient client, MqttClientPersistence persistence, M
 
 		this.tokenStore = new CommsTokenStore(getClient().getClientId());
 		this.callback 	= new CommsCallback(this);
-		this.clientState = new ClientState(persistence, tokenStore, this.callback, this, pingSender);
+		this.clientState = new ClientState(persistence, tokenStore, this.callback, this, pingSender, highResolutionTimer);
 
 		callback.setClientState(clientState);
 		log.setResourceName(getClient().getClientId());

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

@@ -33,6 +33,8 @@
 import org.eclipse.paho.client.mqttv3.internal.ConnectActionListener;
 import org.eclipse.paho.client.mqttv3.internal.DisconnectedMessageBuffer;
 import org.eclipse.paho.client.mqttv3.internal.ExceptionHelper;
+import org.eclipse.paho.client.mqttv3.internal.HighResolutionTimer;
+import org.eclipse.paho.client.mqttv3.internal.SystemHighResolutionTimer;
 import org.eclipse.paho.client.mqttv3.internal.NetworkModule;
 import org.eclipse.paho.client.mqttv3.internal.NetworkModuleService;
 import org.eclipse.paho.client.mqttv3.internal.wire.MqttDisconnect;
@@ -433,6 +435,124 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence
 	 */
 	public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence persistence,
 			MqttPingSender pingSender, ScheduledExecutorService executorService) throws MqttException {
+		this(serverURI, clientId, persistence, pingSender, executorService, null);
+	}
+	/**
+	 * Create an MqttAsyncClient that is used to communicate with an MQTT
+	 * server.
+	 * <p>
+	 * The address of a server can be specified on the constructor.
+	 * Alternatively a list containing one or more servers can be specified
+	 * using the {@link MqttConnectOptions#setServerURIs(String[])
+	 * setServerURIs} method on MqttConnectOptions.
+	 *
+	 * <p>
+	 * The <code>serverURI</code> parameter is typically used with the the
+	 * <code>clientId</code> parameter to form a key. The key is used to store
+	 * and reference messages while they are being delivered. Hence the
+	 * serverURI specified on the constructor must still be specified even if a
+	 * list of servers is specified on an MqttConnectOptions object. The
+	 * serverURI on the constructor must remain the same across restarts of the
+	 * client for delivery of messages to be maintained from a given client to a
+	 * given server or set of servers.
+	 *
+	 * <p>
+	 * The address of the server to connect to is specified as a URI. Two types
+	 * of connection are supported <code>tcp://</code> for a TCP connection and
+	 * <code>ssl://</code> for a TCP connection secured by SSL/TLS. For example:
+	 * </p>
+	 * <ul>
+	 * <li><code>tcp://localhost:1883</code></li>
+	 * <li><code>ssl://localhost:8883</code></li>
+	 * </ul>
+	 * <p>
+	 * If the port is not specified, it will default to 1883 for
+	 * <code>tcp://</code>" URIs, and 8883 for <code>ssl://</code> URIs.
+	 * </p>
+	 *
+	 * <p>
+	 * A client identifier <code>clientId</code> must be specified and be less
+	 * that 65535 characters. It must be unique across all clients connecting to
+	 * the same server. The clientId is used by the server to store data related
+	 * to the client, hence it is important that the clientId remain the same
+	 * when connecting to a server if durable subscriptions or reliable
+	 * messaging are required.
+	 * <p>
+	 * A convenience method is provided to generate a random client id that
+	 * should satisfy this criteria - {@link #generateClientId()}. As the client
+	 * identifier is used by the server to identify a client when it reconnects,
+	 * the client must use the same identifier between connections if durable
+	 * subscriptions or reliable delivery of messages is required.
+	 * </p>
+	 * <p>
+	 * In Java SE, SSL can be configured in one of several ways, which the
+	 * client will use in the following order:
+	 * </p>
+	 * <ul>
+	 * <li><strong>Supplying an <code>SSLSocketFactory</code></strong> -
+	 * applications can use
+	 * {@link MqttConnectOptions#setSocketFactory(SocketFactory)} to supply a
+	 * factory with the appropriate SSL settings.</li>
+	 * <li><strong>SSL Properties</strong> - applications can supply SSL
+	 * settings as a simple Java Properties using
+	 * {@link MqttConnectOptions#setSSLProperties(Properties)}.</li>
+	 * <li><strong>Use JVM settings</strong> - There are a number of standard
+	 * Java system properties that can be used to configure key and trust
+	 * stores.</li>
+	 * </ul>
+	 *
+	 * <p>
+	 * In Java ME, the platform settings are used for SSL connections.
+	 * </p>
+	 * <p>
+	 * A persistence mechanism is used to enable reliable messaging. For
+	 * messages sent at qualities of service (QoS) 1 or 2 to be reliably
+	 * delivered, messages must be stored (on both the client and server) until
+	 * the delivery of the message is complete. If messages are not safely
+	 * stored when being delivered then a failure in the client or server can
+	 * result in lost messages. A pluggable persistence mechanism is supported
+	 * via the {@link MqttClientPersistence} interface. An implementer of this
+	 * interface that safely stores messages must be specified in order for
+	 * delivery of messages to be reliable. In addition
+	 * {@link MqttConnectOptions#setCleanSession(boolean)} must be set to false.
+	 * In the event that only QoS 0 messages are sent or received or
+	 * cleanSession is set to true then a safe store is not needed.
+	 * </p>
+	 * <p>
+	 * An implementation of file-based persistence is provided in class
+	 * {@link MqttDefaultFilePersistence} which will work in all Java SE based
+	 * systems. If no persistence is needed, the persistence parameter can be
+	 * explicitly set to <code>null</code>.
+	 * </p>
+	 *
+	 * @param serverURI
+	 *            the address of the server to connect to, specified as a URI.
+	 *            Can be overridden using
+	 *            {@link MqttConnectOptions#setServerURIs(String[])}
+	 * @param clientId
+	 *            a client identifier that is unique on the server being
+	 *            connected to
+	 * @param persistence
+	 *            the persistence class to use to store in-flight message. If
+	 *            null then the default persistence mechanism is used
+	 * @param pingSender
+	 *            Custom {@link MqttPingSender} implementation.
+	 * @param executorService
+	 *            used for managing threads. If null no executor service is used.
+	 * @param highResolutionTimer
+	 * 			  used for providing time values for keepalive ping scheduling. If {@code null}, a default timer is used
+	 * @throws IllegalArgumentException
+	 *             if the URI does not start with "tcp://", "ssl://" or
+	 *             "local://"
+	 * @throws IllegalArgumentException
+	 *             if the clientId is null or is greater than 65535 characters
+	 *             in length
+	 * @throws MqttException
+	 *             if any other problem was encountered
+	 */
+	public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence persistence,
+						   MqttPingSender pingSender, ScheduledExecutorService executorService,
+						   HighResolutionTimer highResolutionTimer) throws MqttException {
 		final String methodName = "MqttAsyncClient";
 
 		log.setResourceName(clientId);
@@ -461,13 +581,17 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence
 			this.persistence = new MemoryPersistence();
 		}
 
+		if (highResolutionTimer == null) {
+			highResolutionTimer = new SystemHighResolutionTimer();
+		}
+
 		this.executorService = executorService;
 
 		// @TRACE 101=<init> ClientID={0} ServerURI={1} PersistenceType={2}
 		log.fine(CLASS_NAME, methodName, "101", new Object[] { clientId, serverURI, persistence });
 
 		this.persistence.open(clientId, serverURI);
-		this.comms = new ClientComms(this, this.persistence, pingSender, this.executorService);
+		this.comms = new ClientComms(this, this.persistence, pingSender, this.executorService, highResolutionTimer);
 		this.persistence.close();
 		this.topics = new Hashtable();
 

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

@@ -122,6 +122,7 @@ public class ClientState {
 	private long keepAliveNanos;					// nanoseconds time
 	private boolean cleanSession;
 	private MqttClientPersistence persistence;
+	private HighResolutionTimer highResolutionTimer;
 
 	private int maxInflight = 0;	
 	private int actualInFlight = 0;
@@ -148,7 +149,8 @@ public class ClientState {
 	private MqttPingSender pingSender = null;
 
 	protected ClientState(MqttClientPersistence persistence, CommsTokenStore tokenStore, 
-			CommsCallback callback, ClientComms clientComms, MqttPingSender pingSender) throws MqttException {
+			CommsCallback callback, ClientComms clientComms, MqttPingSender pingSender,
+		    HighResolutionTimer highResolutionTimer) throws MqttException {
 
 		log.setResourceName(clientComms.getClient().getClientId());
 		log.finer(CLASS_NAME, "<Init>", "" );
@@ -168,6 +170,7 @@ protected ClientState(MqttClientPersistence persistence, CommsTokenStore tokenSt
 		this.tokenStore = tokenStore;
 		this.clientComms = clientComms;
 		this.pingSender = pingSender;
+		this.highResolutionTimer = highResolutionTimer;
 
 		restoreState();
 	}
@@ -719,7 +722,7 @@ public MqttToken checkForActivity(IMqttActionListener pingCallback) throws MqttE
 		long nextPingTime = TimeUnit.NANOSECONDS.toMillis(this.keepAliveNanos);		// milliseconds relative time
 
 		if (connected && this.keepAliveNanos > 0) {
-			long time = System.nanoTime();
+			long time = highResolutionTimer.nanoTime();
 			// 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.
@@ -897,7 +900,7 @@ public void setKeepAliveInterval(long interval) {
     public void notifySentBytes(int sentBytesCount) {
         final String methodName = "notifySentBytes";
         if (sentBytesCount > 0) {
-        	this.lastOutboundActivity = System.nanoTime();
+        	this.lastOutboundActivity = highResolutionTimer.nanoTime();
         }
         // @TRACE 643=sent bytes count={0}                                                                                                                                                                                            
         log.fine(CLASS_NAME, methodName, "643", new Object[] {
@@ -912,7 +915,7 @@ public void notifySentBytes(int sentBytesCount) {
 	protected void notifySent(MqttWireMessage message) {
 		final String methodName = "notifySent";
 
-		this.lastOutboundActivity = System.nanoTime();
+		this.lastOutboundActivity = highResolutionTimer.nanoTime();
 		//@TRACE 625=key={0}
 		log.fine(CLASS_NAME,methodName,"625",new Object[]{message.getKey()});
 
@@ -924,7 +927,7 @@ protected void notifySent(MqttWireMessage message) {
 		token.internalTok.notifySent();
         if (message instanceof MqttPingReq) {
             synchronized (pingOutstandingLock) {
-            	long time = System.nanoTime();
+            	long time = highResolutionTimer.nanoTime();
                 synchronized (pingOutstandingLock) {
                 	lastPing = time;
                 	pingOutstanding++;
@@ -978,7 +981,7 @@ protected boolean checkQuiesceLock() {
     public void notifyReceivedBytes(int receivedBytesCount) {
         final String methodName = "notifyReceivedBytes";
         if (receivedBytesCount > 0) {
-            this.lastInboundActivity = System.nanoTime();
+            this.lastInboundActivity = highResolutionTimer.nanoTime();
         }
         // @TRACE 630=received bytes count={0}                                                                                                                                                                                        
         log.fine(CLASS_NAME, methodName, "630", new Object[] {
@@ -993,7 +996,7 @@ public void notifyReceivedBytes(int receivedBytesCount) {
 	 */
 	protected void notifyReceivedAck(MqttAck ack) throws MqttException {
 		final String methodName = "notifyReceivedAck";
-		this.lastInboundActivity = System.nanoTime();
+		this.lastInboundActivity = highResolutionTimer.nanoTime();
 
 		// @TRACE 627=received key={0} message={1}
 		log.fine(CLASS_NAME, methodName, "627", new Object[] {
@@ -1075,7 +1078,7 @@ protected void notifyReceivedAck(MqttAck ack) throws MqttException {
 	 */
 	protected void notifyReceivedMsg(MqttWireMessage message) throws MqttException {
 		final String methodName = "notifyReceivedMsg";
-		this.lastInboundActivity = System.nanoTime();
+		this.lastInboundActivity = highResolutionTimer.nanoTime();
 
 		// @TRACE 651=received key={0} message={1}
 		log.fine(CLASS_NAME, methodName, "651", new Object[] {
@@ -1435,7 +1438,8 @@ protected void close() {
 		callback = null;
 		clientComms = null;
 		persistence = null;
-		pingCommand = null;	
+		pingCommand = null;
+		highResolutionTimer = null;
 	}
 
 	public Properties getDebug() {

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

@@ -0,0 +1,35 @@
+/*******************************************************************************
+ * 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 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:
+ *    Dustin Thomson - initial API and implementation and/or initial documentation
+ */
+
+package org.eclipse.paho.client.mqttv3.internal;
+
+/**
+ * A high-resolution timer source.
+ *
+ * Implementations must use clocks that are guaranteed to be monotonic and continue to run
+ * even if the CPU enters a low-power state.
+ */
+public interface HighResolutionTimer {
+
+    /**
+     * Returns the current value of a high-resolution time source, in nanoseconds.
+     *
+     * This method can only be used to measure elapsed time and may return negative values.
+     *
+     * @return the current value of a high-resolution time source, in nanoseconds.
+     */
+    public long nanoTime();
+}

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

@@ -0,0 +1,32 @@
+/*******************************************************************************
+ * 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 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:
+ *    Dustin Thomson - initial API and implementation and/or initial documentation
+ */
+
+package org.eclipse.paho.client.mqttv3.internal;
+
+/**
+ * A high resolution timer appropriate for use by most JVMs.
+ *
+ * This implementation delegates {@link #nanoTime()} to {@link System#nanoTime()}.
+ *
+ * Note: This implementation is not appropriate for use on Android, as the clock backing {@link System#nanoTime()} stops
+ *       when the system enters deep sleep.
+ */
+public class SystemHighResolutionTimer implements HighResolutionTimer {
+    @Override
+    public long nanoTime() {
+        return System.nanoTime();
+    }
+}