Refactored thread stopping and starting, to not introduce 100ms delays by waiting for threads to start & stop with Thread.sleep(). Threads will enter the running state immediately when start() is called. If the ExecutorService is used, gracefully terminate threads by waiting for their completion, rather than interupting.

Related commits: a6faf77, 4b23f12

Signed-off-by: Liu Woon Yung <pirorin187@gmail.com>

Author: sp193

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

@@ -21,6 +21,7 @@
 import java.util.Enumeration;
 import java.util.Hashtable;
 import java.util.Vector;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 
@@ -98,37 +99,33 @@ public void start(String threadName, ExecutorService executorService) {
 				completeQueue.clear();
 
 				target_state = State.RUNNING;
+				current_state = State.RUNNING;
 				if (executorService == null) {
-					new Thread(this).start();
+					callbackFuture = null;
+					callbackThread = new Thread(this);
+					callbackThread.start();
 				} else {
+					callbackThread = null;
 					callbackFuture = executorService.submit(this);
 				}
 			}
 		}
-		while (!isRunning()) {
-			try { Thread.sleep(100); } catch (Exception e) { }
-		}			
 	}
-	
 
 	/**
 	 * Stops the callback thread. 
 	 * This call will block until stop has completed.
 	 */
 	public void stop() {
 		final String methodName = "stop";
-		
-		synchronized (lifecycle) {
-			if (callbackFuture != null) {
-				callbackFuture.cancel(true);
-			}
-		}
+
 		if (isRunning()) {
 			// @TRACE 700=stopping
 			log.fine(CLASS_NAME, methodName, "700");
 			synchronized (lifecycle) {
 				target_state = State.STOPPED;
 			}
+			// Do not allow a thread to wait for itself.
 			if (!Thread.currentThread().equals(callbackThread)) {
 				synchronized (workAvailable) {
 					// @TRACE 701=notify workAvailable and wait for run
@@ -137,9 +134,16 @@ public void stop() {
 					workAvailable.notifyAll();
 				}
 				// Wait for the thread to finish.
-				while (isRunning()) {
-					try { Thread.sleep(100); } catch (Exception e) { }
-					clientState.notifyQueueLock();
+				if (callbackFuture != null) {
+					try {
+						callbackFuture.get();
+					} catch (ExecutionException | InterruptedException e) {
+					}
+				} else {
+					try {
+						callbackThread.join();
+					} catch (InterruptedException e) {
+					}
 				}
 			}
 			// @TRACE 703=stopped
@@ -163,10 +167,6 @@ public void run() {
 		final String methodName = "run";
 		callbackThread = Thread.currentThread();
 		callbackThread.setName(threadName);
-		
-		synchronized (lifecycle) {
-			current_state = State.RUNNING;
-		}
 
 		while (isRunning()) {
 			try {
@@ -240,7 +240,6 @@ public void run() {
 		synchronized (lifecycle) {
 			current_state = State.STOPPED;
 		}
-		callbackThread = null;
 	}
 
 	private void handleActionComplete(MqttToken token)

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

@@ -17,6 +17,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 
@@ -44,7 +45,7 @@ private enum State {STOPPED, RUNNING, STARTING, RECEIVING};
 	private Object lifecycle = new Object();
 	private String threadName;
 	private Future<?> receiverFuture;
-	
+
 	private ClientState clientState = null;
 	private ClientComms clientComms = null;
 	private MqttInputStream in;
@@ -72,35 +73,47 @@ public void start(String threadName, ExecutorService executorService) {
 		synchronized (lifecycle) {
 			if (current_state == State.STOPPED && target_state == State.STOPPED) {
 				target_state = State.RUNNING;
+				current_state = State.RUNNING;
 				if (executorService == null) {
-					new Thread(this).start();
+					receiverFuture = null;
+					recThread = new Thread(this);
+					recThread.start();
 				} else {
+					recThread = null;
 					receiverFuture = executorService.submit(this);
 				}
 			}
 		}
-		while (!isRunning()) {
-			try { Thread.sleep(100); } catch (Exception e) { }
-		}
 	}
 
 	/**
 	 * Stops the Receiver's thread.  This call will block.
 	 */
 	public void stop() {
 		final String methodName = "stop";
+		boolean isRunning;
+
 		synchronized (lifecycle) {
-			if (receiverFuture != null) {
-				receiverFuture.cancel(true);
-			}
 			//@TRACE 850=stopping
 			log.fine(CLASS_NAME,methodName, "850");
-			if (isRunning()) {
+			isRunning = isRunning();
+			if (isRunning) {
 				target_state = State.STOPPED;
 			}
 		}
-		while (isRunning()) {
-			try { Thread.sleep(100); } catch (Exception e) { }
+		// This and the clause above will prevent a thread from waiting for itself.
+		if (isRunning) {
+			if (receiverFuture != null) {
+				try {
+					receiverFuture.get();
+				} catch (ExecutionException | InterruptedException e) {
+				}
+			} else {
+				try {
+					recThread.join();
+				} catch (InterruptedException e) {
+				}
+			}
 		}
 		//@TRACE 851=stopped
 		log.fine(CLASS_NAME,methodName,"851");
@@ -110,15 +123,10 @@ public void stop() {
 	 * Run loop to receive messages from the server.
 	 */
 	public void run() {
-		recThread = Thread.currentThread();
-		recThread.setName(threadName);
+		Thread.currentThread().setName(threadName);
 		final String methodName = "run";
 		MqttToken token = null;
 
-		synchronized (lifecycle) {
-			current_state = State.RUNNING;
-		}
-		
 		try {
 			State my_target;
 			synchronized (lifecycle) {
@@ -204,7 +212,6 @@ public void run() {
 			}
 		} // end try
 
-		recThread = null;
 		//@TRACE 854=<
 		log.fine(CLASS_NAME,methodName,"854");
 	}

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

@@ -17,6 +17,7 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 
@@ -42,7 +43,7 @@ private enum State {STOPPED, RUNNING, STARTING};
 	private Thread 	sendThread		= null;
 	private String threadName;
 	private Future<?> senderFuture;
-	
+
 	private ClientState clientState = null;
 	private MqttOutputStream out;
 	private ClientComms clientComms = null;
@@ -67,56 +68,62 @@ public void start(String threadName, ExecutorService executorService) {
 		synchronized (lifecycle) {
 			if (current_state == State.STOPPED && target_state == State.STOPPED) {
 				target_state = State.RUNNING;
+				current_state = State.RUNNING;
 				if (executorService == null) {
-					new Thread(this).start();
+					senderFuture = null;
+					sendThread = new Thread(this);
+					sendThread.start();
 				} else {
+					sendThread = null;
 					senderFuture = executorService.submit(this);
 				}
 			}
 		}
-		while (!isRunning()) {
-			try { Thread.sleep(100); } catch (Exception e) { }
-		}
 	}
 
 	/**
 	 * Stops the Sender's thread.  This call will block.
 	 */
 	public void stop() {
 		final String methodName = "stop";
-		
+		boolean isRunning;
+
 		if (!isRunning()) {
 			return;
 		}
-			
+
 		synchronized (lifecycle) {
-			if (senderFuture != null) {
-				senderFuture.cancel(true);
-			}
 			//@TRACE 800=stopping sender
 			log.fine(CLASS_NAME,methodName,"800");
-			if (isRunning()) {
+			isRunning = isRunning();
+			if (isRunning) {
 				target_state = State.STOPPED;
 				clientState.notifyQueueLock();
 			}
 		}
-		while (isRunning()) {
-			try { Thread.sleep(100); } catch (Exception e) { }
-			clientState.notifyQueueLock();
+		// This and the clause above will prevent a thread from waiting for itself.
+		if (isRunning) {
+			if (senderFuture != null) {
+				try {
+					senderFuture.get();
+				} catch (ExecutionException | InterruptedException e) {
+				}
+			} else {
+				try {
+					sendThread.join();
+				} catch (InterruptedException e) {
+					e.printStackTrace();
+				}
+			}
 		}
 		//@TRACE 801=stopped
 		log.fine(CLASS_NAME,methodName,"801");
 	}
 
 	public void run() {
-		sendThread = Thread.currentThread();
-		sendThread.setName(threadName);
+		Thread.currentThread().setName(threadName);
 		final String methodName = "run";
 		MqttWireMessage message = null;
-		
-		synchronized (lifecycle) {
-			current_state = State.RUNNING;
-		}
 
 		try {
 			State my_target;
@@ -176,7 +183,6 @@ public void run() {
 		} finally {
 			synchronized (lifecycle) {
 				current_state = State.STOPPED;
-				sendThread = null;
 			}
 		}