Refactor thread stopping and starting #207

Author: Ian Craggs

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

@@ -23,7 +23,6 @@
 import java.util.Vector;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
-import java.util.concurrent.Semaphore;
 
 import org.eclipse.paho.client.mqttv3.IMqttActionListener;
 import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
@@ -37,6 +36,7 @@
 import org.eclipse.paho.client.mqttv3.internal.wire.MqttPubAck;
 import org.eclipse.paho.client.mqttv3.internal.wire.MqttPubComp;
 import org.eclipse.paho.client.mqttv3.internal.wire.MqttPublish;
+import org.eclipse.paho.client.mqttv3.internal.wire.MqttWireMessage;
 import org.eclipse.paho.client.mqttv3.logging.Logger;
 import org.eclipse.paho.client.mqttv3.logging.LoggerFactory;
 
@@ -52,27 +52,29 @@ public class CommsCallback implements Runnable {
 	private static final int INBOUND_QUEUE_SIZE = 10;
 	private MqttCallback mqttCallback;
 	private MqttCallbackExtended reconnectInternalCallback;
-	private Hashtable callbacks; // topicFilter -> messageHandler
+	private Hashtable<String, IMqttMessageListener> callbacks; // topicFilter -> messageHandler
 	private ClientComms clientComms;
-	private Vector messageQueue;
-	private Vector completeQueue;
-	public boolean running = false;
-	private boolean quiescing = false;
+	private Vector<MqttWireMessage> messageQueue;
+	private Vector<MqttToken> completeQueue;
+	
+	private enum State {STOPPED, RUNNING, QUIESCING};
+	private State current_state = State.STOPPED;
+	private State target_state = State.STOPPED;
 	private Object lifecycle = new Object();
 	private Thread callbackThread;
+	private String threadName;
+	private Future<?> callbackFuture;
+	
 	private Object workAvailable = new Object();
 	private Object spaceAvailable = new Object();
 	private ClientState clientState;
 	private boolean manualAcks = false;
-	private String threadName;
-	private final Semaphore runningSemaphore = new Semaphore(1);
-	private Future callbackFuture;
 
 	CommsCallback(ClientComms clientComms) {
 		this.clientComms = clientComms;
-		this.messageQueue = new Vector(INBOUND_QUEUE_SIZE);
-		this.completeQueue = new Vector(INBOUND_QUEUE_SIZE);
-		this.callbacks = new Hashtable();
+		this.messageQueue = new Vector<MqttWireMessage>(INBOUND_QUEUE_SIZE);
+		this.completeQueue = new Vector<MqttToken>(INBOUND_QUEUE_SIZE);
+		this.callbacks = new Hashtable<String, IMqttMessageListener>();
 		log.setResourceName(clientComms.getClient().getClientId());
 	}
 
@@ -87,55 +89,59 @@ public void setClientState(ClientState clientState) {
 	 */
 	public void start(String threadName, ExecutorService executorService) {
 		this.threadName = threadName;
+
 		synchronized (lifecycle) {
-			if (!running) {
+			if (current_state == State.STOPPED) {
 				// Preparatory work before starting the background thread.
 				// For safety ensure any old events are cleared.
 				messageQueue.clear();
 				completeQueue.clear();
-
-				running = true;
-				quiescing = false;
+				
+				target_state = State.RUNNING;
 				if (executorService == null) {
 					new Thread(this).start();
 				} else {
 					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 (running) {
-				// @TRACE 700=stopping
-				log.fine(CLASS_NAME, methodName, "700");
-				running = false;
-				if (!Thread.currentThread().equals(callbackThread)) {
-					try {
-						synchronized (workAvailable) {
-							// @TRACE 701=notify workAvailable and wait for run
-							// to finish
-							log.fine(CLASS_NAME, methodName, "701");
-							workAvailable.notifyAll();
-						}
-						// Wait for the thread to finish.
-						runningSemaphore.acquire();
-					} catch (InterruptedException ex) {
-					} finally {
-						runningSemaphore.release();
-					}
+		}
+		if (isRunning()) {
+			// @TRACE 700=stopping
+			log.fine(CLASS_NAME, methodName, "700");
+			synchronized (lifecycle) {
+				target_state = State.STOPPED;
+			}
+			if (!Thread.currentThread().equals(callbackThread)) {
+				synchronized (workAvailable) {
+					// @TRACE 701=notify workAvailable and wait for run
+					// to finish
+					log.fine(CLASS_NAME, methodName, "701");
+					workAvailable.notifyAll();
+				}
+				// Wait for the thread to finish.
+				while (isRunning()) {
+					try { Thread.sleep(100); } catch (Exception e) { }
+					clientState.notifyQueueLock();
 				}
 			}
-			callbackThread = null;
 			// @TRACE 703=stopped
 			log.fine(CLASS_NAME, methodName, "703");
 		}
@@ -157,20 +163,17 @@ public void run() {
 		final String methodName = "run";
 		callbackThread = Thread.currentThread();
 		callbackThread.setName(threadName);
-
-		try {
-			runningSemaphore.acquire();
-		} catch (InterruptedException e) {
-			running = false;
-			return;
+		
+		synchronized (lifecycle) {
+			current_state = State.RUNNING;
 		}
 
-		while (running) {
+		while (isRunning()) {
 			try {
 				// If no work is currently available, then wait until there is some...
 				try {
 					synchronized (workAvailable) {
-						if (running && messageQueue.isEmpty()
+						if (isRunning() && messageQueue.isEmpty()
 								&& completeQueue.isEmpty()) {
 							// @TRACE 704=wait for workAvailable
 							log.fine(CLASS_NAME, methodName, "704");
@@ -180,7 +183,7 @@ public void run() {
 				} catch (InterruptedException e) {
 				}
 
-				if (running) {
+				if (isRunning()) {
 					// Check for deliveryComplete callbacks...
 					MqttToken token = null;
 					synchronized (completeQueue) {
@@ -211,7 +214,7 @@ public void run() {
 					}
 				}
 
-				if (quiescing) {
+				if (isQuiescing()) {
 					clientState.checkQuiesceLock();
 				}
 
@@ -220,10 +223,10 @@ public void run() {
 				// of class NoClassDefFoundError
 				// @TRACE 714=callback threw exception
 				log.fine(CLASS_NAME, methodName, "714", null, ex);
-				running = false;
+
 				clientComms.shutdownConnection(null, new MqttException(ex));
 			} finally {
-				runningSemaphore.release();
+
 			    synchronized (spaceAvailable) {
                     // Notify the spaceAvailable lock, to say that there's now
                     // some space on the queue...
@@ -234,6 +237,10 @@ public void run() {
                 }
 			}
 		}
+		synchronized (lifecycle) {
+			current_state = State.STOPPED;
+		}
+		callbackThread = null;
 	}
 
 	private void handleActionComplete(MqttToken token)
@@ -349,7 +356,7 @@ public void messageArrived(MqttPublish sendMessage) {
 			// the client protect itself from getting flooded by messages 
 			// from the server.
 			synchronized (spaceAvailable) {
-				while (running && !quiescing && messageQueue.size() >= INBOUND_QUEUE_SIZE) {
+				while (isRunning() && !isQuiescing() && messageQueue.size() >= INBOUND_QUEUE_SIZE) {
 					try {
 						// @TRACE 709=wait for spaceAvailable
 						log.fine(CLASS_NAME, methodName, "709");
@@ -358,7 +365,7 @@ public void messageArrived(MqttPublish sendMessage) {
 					}
 				}
 			}
-			if (!quiescing) {
+			if (!isQuiescing()) {
 				messageQueue.addElement(sendMessage);
 				// Notify the CommsCallback thread that there's work to do...
 				synchronized (workAvailable) {
@@ -377,7 +384,10 @@ public void messageArrived(MqttPublish sendMessage) {
 	 */
 	public void quiesce() {
 		final String methodName = "quiesce";
-		this.quiescing = true;
+		synchronized (lifecycle) {
+			if (current_state == State.RUNNING)
+			current_state = State.QUIESCING;
+		}
 		synchronized (spaceAvailable) {
 			// @TRACE 711=quiesce notify spaceAvailable
 			log.fine(CLASS_NAME, methodName, "711");
@@ -387,7 +397,7 @@ public void quiesce() {
 	}
 
 	public boolean isQuiesced() {
-		if (quiescing && completeQueue.size() == 0 && messageQueue.size() == 0) {
+		if (isQuiescing() && completeQueue.size() == 0 && messageQueue.size() == 0) {
 			return true;
 		}
 		return false;
@@ -435,7 +445,7 @@ public void messageArrivedComplete(int messageId, int qos)
 	public void asyncOperationComplete(MqttToken token) {
 		final String methodName = "asyncOperationComplete";
 
-		if (running) {
+		if (isRunning()) {
 			// invoke callbacks on callback thread
 			completeQueue.addElement(token);
 			synchronized (workAvailable) {
@@ -487,7 +497,7 @@ protected boolean deliverMessage(String topicName, int messageId, MqttMessage aM
 	{		
 		boolean delivered = false;
 
-		Enumeration keys = callbacks.keys();
+		Enumeration<String> keys = callbacks.keys();
 		while (keys.hasMoreElements()) {
 			String topicFilter = (String)keys.nextElement();
 			if (MqttTopic.isMatched(topicFilter, topicName)) {
@@ -506,5 +516,22 @@ protected boolean deliverMessage(String topicName, int messageId, MqttMessage aM
 
 		return delivered;
 	}
+	
+	public boolean isRunning() {
+		boolean result;
+		synchronized (lifecycle) {
+			result = ((current_state == State.RUNNING || current_state == State.QUIESCING)
+					&& target_state == State.RUNNING);
+		}
+		return result;
+	}
+	
+	public boolean isQuiescing() {
+		boolean result;
+		synchronized (lifecycle) {
+			result = (current_state == State.QUIESCING);
+		}
+		return result;
+	}
 
 }