Merging in Java 1.7+ Changes

* Use random UUID for 16-byte key in handshake request
Signed-off-by: miketran <miketran@us.ibm.com>

* Add Server Name Indication support - Java 7 is required
Signed-off-by: miketran78727 <miketran@us.ibm.com>

* Use ExecutorService for managing threads (#280)
Change-Id: Ie3996a663f2870d1baa8910bed276bae37e1e369
Signed-off-by: David Katz <David.Katz@bmw-carit.de>

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

Author: jpwsutton

org.eclipse.paho.client.mqttv3.test/pom.xml

@@ -51,6 +51,7 @@
         			<forkCount>1</forkCount>
 					<systemPropertyVariables>
 						<SERVER_URI>${test.server_uri}</SERVER_URI>
+						<SERVER_SSL_PORT>${test.server_ssl_port}</SERVER_SSL_PORT>
 					</systemPropertyVariables>
 					<excludes>
 						<exclude>${test.exclude}</exclude>

org.eclipse.paho.client.mqttv3/build.xml

@@ -16,8 +16,8 @@
 	<property name="build.level" value="LYYMMDD" />
 
 	<!-- these should be moved to properties file if required -->
-	<property name="javac_target" value="1.2" />
-	<property name="javac_source" value="1.2" />
+	<property name="javac_target" value="1.7" />
+	<property name="javac_source" value="1.7" />
 	<property name="javac_debug" value="on" />
 	<property name="javac_debuglevel" value="lines,vars,source" />
 

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

@@ -22,6 +22,8 @@
 import java.util.Enumeration;
 import java.util.Properties;
 import java.util.Vector;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 
 import org.eclipse.paho.client.mqttv3.BufferedMessage;
 import org.eclipse.paho.client.mqttv3.IMqttActionListener;
@@ -80,7 +82,9 @@ public class ClientComms {
 	private boolean	closePending = false;
 	private boolean resting = false;
 	private DisconnectedMessageBuffer disconnectedMessageBuffer;
-	
+
+	private ExecutorService executorService;
+
 	/**
 	 * Creates a new ClientComms object, using the specified module to handle
 	 * the network calls.
@@ -89,13 +93,14 @@ public class ClientComms {
 	 * @param pingSender the {@link MqttPingSender}
 	 * @throws MqttException if an exception occurs whilst communicating with the server
 	 */
-	public ClientComms(IMqttAsyncClient client, MqttClientPersistence persistence, MqttPingSender pingSender) throws MqttException {
+	public ClientComms(IMqttAsyncClient client, MqttClientPersistence persistence, MqttPingSender pingSender, ExecutorService executorService) throws MqttException {
 		this.conState = DISCONNECTED;
 		this.client 	= client;
 		this.persistence = persistence;
 		this.pingSender = pingSender;
 		this.pingSender.init(this);
-		
+		this.executorService = executorService;
+
 		this.tokenStore = new CommsTokenStore(getClient().getClientId());
 		this.callback 	= new CommsCallback(this);
 		this.clientState = new ClientState(persistence, tokenStore, this.callback, this, pingSender);
@@ -108,6 +113,22 @@ CommsReceiver getReceiver() {
 		return receiver;
 	}
 
+	private void shutdownExecutorService() {
+		String methodName = "shutdownExecutorService";
+		executorService.shutdown();
+		try {
+			if (!executorService.awaitTermination(1, TimeUnit.SECONDS)) {
+				executorService.shutdownNow();
+				if (!executorService.awaitTermination(1, TimeUnit.SECONDS)) {
+					log.fine(CLASS_NAME, methodName, "executorService did not terminate");
+				}
+			}
+		} catch (InterruptedException ie) {
+			executorService.shutdownNow();
+			Thread.currentThread().interrupt();
+		}
+	}
+
 	/**
 	 * Sends a message to the server. Does not check if connected this validation must be done
 	 * by invoking routines.
@@ -205,7 +226,7 @@ public void close() throws MqttException {
 				}
 
 				conState = CLOSED;
-
+				shutdownExecutorService();
 				// ShutdownConnection has already cleaned most things
 				clientState.close();
 				clientState = null;
@@ -254,7 +275,7 @@ public void connect(MqttConnectOptions options, MqttToken token) throws MqttExce
                 this.clientState.setMaxInflight(conOptions.getMaxInflight());
 
 				tokenStore.open();
-				ConnectBG conbg = new ConnectBG(this, token, connect);
+				ConnectBG conbg = new ConnectBG(this, token, connect, executorService);
 				conbg.start();
 			}
 			else {
@@ -479,7 +500,7 @@ public void disconnect(MqttDisconnect disconnect, long quiesceTimeout, MqttToken
 			//@TRACE 218=state=DISCONNECTING
 			log.fine(CLASS_NAME,methodName,"218");
 			conState = DISCONNECTING;
-			DisconnectBG discbg = new DisconnectBG(disconnect,quiesceTimeout,token);
+			DisconnectBG discbg = new DisconnectBG(disconnect,quiesceTimeout,token, executorService);
 			discbg.start();
 		}
 	}
@@ -636,22 +657,23 @@ public Properties getDebug() {
 	// the socket could take time to create.
 	private class ConnectBG implements Runnable {
 		ClientComms 	clientComms = null;
-		Thread 			cBg = null;
 		MqttToken 		conToken;
 		MqttConnect 	conPacket;
+		private String threadName;
 
-		ConnectBG(ClientComms cc, MqttToken cToken, MqttConnect cPacket) {
+		ConnectBG(ClientComms cc, MqttToken cToken, MqttConnect cPacket, ExecutorService executorService) {
 			clientComms = cc;
 			conToken 	= cToken;
 			conPacket 	= cPacket;
-			cBg = new Thread(this, "MQTT Con: "+getClient().getClientId());
+			threadName = "MQTT Con: "+getClient().getClientId();
 		}
 
 		void start() {
-			cBg.start();
+			executorService.execute(this);
 		}
 
 		public void run() {
+			Thread.currentThread().setName(threadName);
 			final String methodName = "connectBG:run";
 			MqttException mqttEx = null;
 			//@TRACE 220=>
@@ -675,10 +697,10 @@ public void run() {
 				NetworkModule networkModule = networkModules[networkModuleIndex];
 				networkModule.start();
 				receiver = new CommsReceiver(clientComms, clientState, tokenStore, networkModule.getInputStream());
-				receiver.start("MQTT Rec: "+getClient().getClientId());
+				receiver.start("MQTT Rec: "+getClient().getClientId(), executorService);
 				sender = new CommsSender(clientComms, clientState, tokenStore, networkModule.getOutputStream());
-				sender.start("MQTT Snd: "+getClient().getClientId());
-				callback.start("MQTT Call: "+getClient().getClientId());				
+				sender.start("MQTT Snd: "+getClient().getClientId(), executorService);
+				callback.start("MQTT Call: "+getClient().getClientId(), executorService);
 				internalSend(conPacket, conToken);
 			} catch (MqttException ex) {
 				//@TRACE 212=connect failed: unexpected exception
@@ -699,23 +721,24 @@ public void run() {
 	// Kick off the disconnect processing in the background so that it does not block. For instance
 	// the quiesce
 	private class DisconnectBG implements Runnable {
-		Thread dBg = null;
 		MqttDisconnect disconnect;
 		long quiesceTimeout;
 		MqttToken token;
+		private String threadName;
 
-		DisconnectBG(MqttDisconnect disconnect, long quiesceTimeout, MqttToken token ) {
+		DisconnectBG(MqttDisconnect disconnect, long quiesceTimeout, MqttToken token, ExecutorService executorService) {
 			this.disconnect = disconnect;
 			this.quiesceTimeout = quiesceTimeout;
 			this.token = token;
 		}
 
 		void start() {
-			dBg = new Thread(this, "MQTT Disc: "+getClient().getClientId());
-			dBg.start();
+			threadName = "MQTT Disc: "+getClient().getClientId();
+			executorService.execute(this);
 		}
-		
+
 		public void run() {
+			Thread.currentThread().setName(threadName);
 			final String methodName = "disconnectBG:run";
 			//@TRACE 221=>
 			log.fine(CLASS_NAME, methodName, "221");
@@ -835,7 +858,7 @@ public void publishBufferedMessage(BufferedMessage bufferedMessage) throws MqttE
 					}	
 				}
 			});
-			new Thread(disconnectedMessageBuffer).start();
+			executorService.execute(disconnectedMessageBuffer);
 		}
 	}