Merge branch 'develop' of github.com:eclipse/paho.mqtt.java into develop

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

Author: jpwsutton

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

@@ -104,7 +104,7 @@ public class MqttAsyncClient implements IMqttAsyncClient {
 	private Timer reconnectTimer; // Automatic reconnect timer
 	private static int reconnectDelay = 1000;  // Reconnect delay, starts at 1 second
 	private boolean reconnecting = false;
-	
+	private static Object clientLock = new Object(); // Simple lock
 
 
 
@@ -1217,21 +1217,32 @@ private void stopReconnectCycle(){
 		String methodName = "stopReconnectCycle";
 		//@Trace 504=Stop reconnect timer for client: {0}
 		log.fine(CLASS_NAME, methodName, "504", new Object[]{this.clientId});
-		if(reconnectTimer != null){
-			reconnectTimer.cancel();
+		synchronized(clientLock) {
+			if (this.connOpts.isAutomaticReconnect()) {
+				if(reconnectTimer != null){
+					reconnectTimer.cancel();
+					reconnectTimer = null;
+				}
+				reconnectDelay = 1000; // Reset Delay Timer
+			}
 		}
-		reconnectDelay = 1000; // Reset Delay Timer
-		
 	}
 
 	private void rescheduleReconnectCycle(int delay){
 		String methodName = "rescheduleReconnectCycle";
 		//@Trace 505=Rescheduling reconnect timer for client: {0}, delay: {1}
 		log.fine(CLASS_NAME, methodName, "505", new Object[]{this.clientId, new Long(reconnectDelay)});
-		if(reconnectTimer != null){
-			reconnectTimer.schedule(new ReconnectTask(), reconnectDelay);
+		synchronized(clientLock) {
+			if(this.connOpts.isAutomaticReconnect()) {
+				if (reconnectTimer != null) {
+					reconnectTimer.schedule(new ReconnectTask(), delay);
+				} else {
+					// The previous reconnect timer was cancelled
+					reconnectDelay = delay;
+					startReconnectCycle();
+				}
+			}
 		}
-
 	}
 
 	private class ReconnectTask extends TimerTask {

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

@@ -500,26 +500,29 @@ public void setServerURIs(String[] array) {
 	 * @param srvURI The Server URI
 	 * @return the URI type
 	 */
-	protected static int validateURI(String srvURI) {
+	public static int validateURI(String srvURI) {
 		try {
 			URI vURI = new URI(srvURI);
-			if (vURI.getScheme().equals("ws")){
+			if ("ws".equals(vURI.getScheme())){
 				return URI_TYPE_WS;
 			}
-			else if (vURI.getScheme().equals("wss")) {
+			else if ("wss".equals(vURI.getScheme())) {
 				return URI_TYPE_WSS;
 			}
 
-			if (!vURI.getPath().equals("")) {
-				throw new IllegalArgumentException(srvURI);
+			if ((vURI.getPath() == null) || vURI.getPath().isEmpty()) {
+				// No op path must be empty
 			}
-			if (vURI.getScheme().equals("tcp")) {
+			else {
+				throw new IllegalArgumentException(srvURI);
+			} 
+			if ("tcp".equals(vURI.getScheme())) {
 				return URI_TYPE_TCP;
 			}
-			else if (vURI.getScheme().equals("ssl")) {
+			else if ("ssl".equals(vURI.getScheme())) {
 				return URI_TYPE_SSL;
 			}
-			else if (vURI.getScheme().equals("local")) {
+			else if ("local".equals(vURI.getScheme())) {
 				return URI_TYPE_LOCAL;
 			}
 			else {

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

@@ -615,6 +615,13 @@ protected void undo(MqttPublish message) throws MqttPersistenceException {
 			pendingMessages.removeElement(message);
 			persistence.remove(getSendPersistenceKey(message));
 			tokenStore.removeToken(message);
+			if(message.getMessage().getQos() > 0){
+				//Free this message Id so it can be used again
+				releaseMessageId(message.getMessageId());
+				//Set the messageId to 0 so if it's ever retried, it will get a new messageId
+				message.setMessageId(0);
+			}
+
 			checkQuiesceLock();
 		}
 	}

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

@@ -3,11 +3,11 @@
  *
  * 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. 
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
  *
- * The Eclipse Public License is available at 
+ * The Eclipse Public License is available at
  *    http://www.eclipse.org/legal/epl-v10.html
- * and the Eclipse Distribution License is available at 
+ * and the Eclipse Distribution License is available at
  *   http://www.eclipse.org/org/documents/edl-v10.php.
  *
  * Contributors:
@@ -44,15 +44,15 @@ public class CommsReceiver implements Runnable {
 	private CommsTokenStore tokenStore = null;
 	private Thread recThread = null;
 	private volatile boolean receiving;
-	
+
 	public CommsReceiver(ClientComms clientComms, ClientState clientState,CommsTokenStore tokenStore, InputStream in) {
 		this.in = new MqttInputStream(clientState, in);
 		this.clientComms = clientComms;
 		this.clientState = clientState;
 		this.tokenStore = tokenStore;
 		log.setResourceName(clientComms.getClient().getClientId());
 	}
-	
+
 	/**
 	 * Starts up the Receiver's thread.
 	 * @param threadName The name of the thread
@@ -100,29 +100,29 @@ public void stop() {
 		//@TRACE 851=stopped
 		log.fine(CLASS_NAME,methodName,"851");
 	}
-	
+
 	/**
 	 * Run loop to receive messages from the server.
 	 */
 	public void run() {
 		final String methodName = "run";
 		MqttToken token = null;
-		
+
 		while (running && (in != null)) {
 			try {
 				//@TRACE 852=network read message
 				log.fine(CLASS_NAME,methodName,"852");
 				receiving = in.available() > 0;
 				MqttWireMessage message = in.readMqttWireMessage();
 				receiving = false;
-				
+
 				// instanceof checks if message is null
 				if (message instanceof MqttAck) {
 					token = tokenStore.getToken(message);
 					if (token!=null) {
 						synchronized (token) {
 							// Ensure the notify processing is done under a lock on the token
-							// This ensures that the send processing can complete  before the 
+							// This ensures that the send processing can complete  before the
 							// receive processing starts! ( request and ack and ack processing
 							// can occur before request processing is complete if not!
 							clientState.notifyReceivedAck((MqttAck)message);
@@ -133,7 +133,6 @@ public void run() {
 						//because of timeouts, crashes, disconnects, restarts etc.
 						//It should be safe to ignore these unexpected messages.
 						log.fine(CLASS_NAME, methodName, "857");
-
 					} else {
 						// It its an ack and there is no token then something is not right.
 						// An ack should always have a token assoicated with it.
@@ -152,13 +151,13 @@ public void run() {
 				running = false;
 				// Token maybe null but that is handled in shutdown
 				clientComms.shutdownConnection(token, ex);
-			} 
+			}
 			catch (IOException ioe) {
 				//@TRACE 853=Stopping due to IOException
 				log.fine(CLASS_NAME,methodName,"853");
 
 				running = false;
-				// An EOFException could be raised if the broker processes the 
+				// An EOFException could be raised if the broker processes the
 				// DISCONNECT and ends the socket before we complete. As such,
 				// only shutdown the connection if we're not already shutting down.
 				if (!clientComms.isDisconnecting()) {
@@ -169,18 +168,18 @@ public void run() {
 				receiving = false;
 			}
 		}
-		
+
 		//@TRACE 854=<
 		log.fine(CLASS_NAME,methodName,"854");
 	}
-	
+
 	public boolean isRunning() {
 		return running;
 	}
-	
+
 	/**
 	 * Returns the receiving state.
-	 * 
+	 *
 	 * @return true if the receiver is receiving data, false otherwise.
 	 */
 	public boolean isReceiving() {

pom.xml

@@ -94,6 +94,21 @@
 						<target>${java.version}</target>
 					</configuration>
 				</plugin>
+				<plugin>
+					<groupId>org.apache.maven.plugins</groupId>
+					<artifactId>maven-javadoc-plugin</artifactId>
+					<version>2.9.1</version>
+				</plugin>
+				<plugin>
+					<groupId>org.apache.maven.plugins</groupId>
+					<artifactId>maven-resources-plugin</artifactId>
+					<version>2.4.2</version>
+				</plugin>
+				<plugin>
+					<groupId>org.apache.maven.plugins</groupId>
+					<artifactId>maven-source-plugin</artifactId>
+					<version>3.0.1</version>
+				</plugin>
 				<plugin>
 					<groupId>org.apache.maven.plugins</groupId>
 					<artifactId>maven-deploy-plugin</artifactId>