Fixing Client Shutdown Bug(#379)

* Do not quiesce if clientState is null

Fixes NPE experienced in some shutdown scenarios

Change-Id: I41b0bfcc64a8e5b60d5a04cc116ad8e2f5cf3fff
Signed-off-by: David Katz <David.Katz@bmw-carit.de>

* Shutdown tasks submitted to the executor in stop methods

Change-Id: I4e1de3bfa8b330b831fff981eeecea3c501b42f5
Signed-off-by: David Katz <David.Katz@bmw-carit.de>

Author: david-katz

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

@@ -519,7 +519,9 @@ public void disconnectForcibly(long quiesceTimeout, long disconnectTimeout) thro
 	 */
 	public void disconnectForcibly(long quiesceTimeout, long disconnectTimeout, boolean sendDisconnectPacket) throws MqttException {
 		// Allow current inbound and outbound work to complete
-		clientState.quiesce(quiesceTimeout);
+		if (clientState != null) {
+			clientState.quiesce(quiesceTimeout);
+		}
 		MqttToken token = new MqttToken(client.getClientId());
 		try {
 			// Send disconnect packet

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

@@ -22,6 +22,7 @@
 import java.util.Hashtable;
 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;
@@ -65,6 +66,7 @@ public class CommsCallback implements Runnable {
 	private boolean manualAcks = false;
 	private String threadName;
 	private final Semaphore runningSemaphore = new Semaphore(1);
+	private Future callbackFuture;
 
 	CommsCallback(ClientComms clientComms) {
 		this.clientComms = clientComms;
@@ -94,7 +96,7 @@ public void start(String threadName, ExecutorService executorService) {
 
 				running = true;
 				quiescing = false;
-				executorService.execute(this);
+				callbackFuture = executorService.submit(this);
 			}
 		}
 	}
@@ -106,6 +108,9 @@ public void start(String threadName, ExecutorService executorService) {
 	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");

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

@@ -18,6 +18,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 import java.util.concurrent.Semaphore;
 
 import org.eclipse.paho.client.mqttv3.MqttException;
@@ -48,6 +49,7 @@ public class CommsReceiver implements Runnable {
 	private volatile boolean receiving;
 	private final Semaphore runningSemaphore = new Semaphore(1);
 	private String threadName;
+	private Future receiverFuture;
 
 
 	public CommsReceiver(ClientComms clientComms, ClientState clientState,CommsTokenStore tokenStore, InputStream in) {
@@ -71,7 +73,7 @@ public void start(String threadName, ExecutorService executorService) {
 		synchronized (lifecycle) {
 			if (!running) {
 				running = true;
-				executorService.execute(this);
+				receiverFuture = executorService.submit(this);
 			}
 		}
 	}
@@ -82,6 +84,9 @@ public void start(String threadName, ExecutorService executorService) {
 	public void stop() {
 		final String methodName = "stop";
 		synchronized (lifecycle) {
+			if (receiverFuture != null) {
+				receiverFuture.cancel(true);
+			}
 			//@TRACE 850=stopping
 			log.fine(CLASS_NAME,methodName, "850");
 			if (running) {

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

@@ -18,6 +18,7 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
@@ -46,6 +47,7 @@ public class CommsSender implements Runnable {
 
 	private String threadName;
 	private final Semaphore runningSemaphore = new Semaphore(1);
+	private Future senderFuture;
 
 	public CommsSender(ClientComms clientComms, ClientState clientState, CommsTokenStore tokenStore, OutputStream out) {
 		this.out = new MqttOutputStream(clientState, out);
@@ -65,7 +67,7 @@ public void start(String threadName, ExecutorService executorService) {
 		synchronized (lifecycle) {
 			if (!running) {
 				running = true;
-				executorService.execute(this);
+				senderFuture = executorService.submit(this);
 			}
 		}
 	}
@@ -77,6 +79,9 @@ public void stop() {
 		final String methodName = "stop";
 
 		synchronized (lifecycle) {
+			if (senderFuture != null) {
+				senderFuture.cancel(true);
+			}
 			//@TRACE 800=stopping sender
 			log.fine(CLASS_NAME,methodName,"800");
 			if (running) {