Fixed MQTT Con thread leak on connection failure #850

Author: PavelAnikeichyk

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

@@ -57,10 +57,10 @@ public class CommsCallback implements Runnable {
 	private final Vector<MqttWireMessage> messageQueue;
 	private final Vector<MqttToken> completeQueue;
 
-	private enum State {STOPPED, RUNNING, QUIESCING}
+	private enum State {READY, RUNNING, QUIESCING, STOPPED}
 
-	private State current_state = State.STOPPED;
-	private State target_state = State.STOPPED;
+	private State current_state = State.READY;
+	private State target_state = State.READY;
 	private final Object lifecycle = new Object();
 	private Thread callbackThread;
 	private String threadName;
@@ -92,7 +92,7 @@ public void start(String threadName, ExecutorService executorService) {
 		this.threadName = threadName;
 
 		synchronized (lifecycle) {
-			if (current_state == State.STOPPED) {
+			if (current_state == State.READY) {
 				// Preparatory work before starting the background thread.
 				// For safety ensure any old events are cleared.
 				messageQueue.clear();
@@ -108,6 +108,9 @@ public void start(String threadName, ExecutorService executorService) {
 		}
 		while (!isRunning()) {
 			try { Thread.sleep(100); } catch (Exception e) { }
+			if (current_state == State.STOPPED) {
+				break;
+			}
 		}			
 	}
 

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

@@ -38,10 +38,10 @@ public class CommsReceiver implements Runnable {
 	private static final String CLASS_NAME = CommsReceiver.class.getName();
 	private Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
 
-	private enum State {STOPPED, RUNNING, STARTING, RECEIVING}
+	private enum State {READY, RUNNING, STARTING, RECEIVING, STOPPED}
 
-	private State current_state = State.STOPPED;
-	private State target_state = State.STOPPED;
+	private State current_state = State.READY;
+	private State target_state = State.READY;
 	private final Object lifecycle = new Object();
 	private String threadName;
 	private Future<?> receiverFuture;
@@ -71,7 +71,7 @@ public void start(String threadName, ExecutorService executorService) {
 		//@TRACE 855=starting
 		log.fine(CLASS_NAME,methodName, "855");
 		synchronized (lifecycle) {
-			if (current_state == State.STOPPED && target_state == State.STOPPED) {
+			if (current_state == State.READY && target_state == State.READY) {
 				target_state = State.RUNNING;
 				if (executorService == null) {
 					new Thread(this).start();
@@ -82,6 +82,9 @@ public void start(String threadName, ExecutorService executorService) {
 		}
 		while (!isRunning()) {
 			try { Thread.sleep(100); } catch (Exception e) { }
+			if (current_state == State.STOPPED) {
+				break;
+			}
 		}
 	}
 

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

@@ -35,10 +35,10 @@ public class CommsSender implements Runnable {
 	private Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
 
 	//Sends MQTT packets to the server on its own thread
-	private enum State {STOPPED, RUNNING, STARTING}
+	private enum State {READY, RUNNING, STARTING, STOPPED}
 
-    private State current_state = State.STOPPED;
-	private State target_state = State.STOPPED;
+    private State current_state = State.READY;
+	private State target_state = State.READY;
 	private final Object lifecycle = new Object();
 	private Thread 	sendThread		= null;
 	private String threadName;
@@ -66,7 +66,7 @@ public CommsSender(ClientComms clientComms, ClientState clientState, CommsTokenS
 	public void start(String threadName, ExecutorService executorService) {
 		this.threadName = threadName;
 		synchronized (lifecycle) {
-			if (current_state == State.STOPPED && target_state == State.STOPPED) {
+			if (current_state == State.READY && target_state == State.READY) {
 				target_state = State.RUNNING;
 				if (executorService == null) {
 					new Thread(this).start();
@@ -77,6 +77,9 @@ public void start(String threadName, ExecutorService executorService) {
 		}
 		while (!isRunning()) {
 			try { Thread.sleep(100); } catch (Exception e) { }
+			if (current_state == State.STOPPED) {
+				break;
+			}
 		}
 	}