Stop threads after disconnect when no executor service #402

Ian Craggs · Ian Craggs · commit 576455ef909a · 2018-12-21T11:26:43.000Z
diff --git a/org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/BasicTest.java b/org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/BasicTest.java
@@ -16,13 +16,15 @@
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.UUID;
+import java.util.concurrent.Executors;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import org.eclipse.paho.client.mqttv3.IMqttClient;
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
@@ -352,6 +354,119 @@ public void test330() throws Exception {
       client.close();
     }
   }
+  
+  
+  @Test
+  public void test402() throws Exception {
+    String methodName = Utility.getMethodName();
+    LoggingUtilities.banner(log, cclass, methodName);
+
+    IMqttClient client = null;
+    int before_thread_count = Thread.activeCount();
+    try {
+      String clientId = methodName;
+      client = clientFactory.createMqttClient(serverURI, clientId);
+
+      log.info("Connecting...(serverURI:" + serverURI + ", ClientId:" + clientId);
+      client.connect();
+
+	  String clientId2 = client.getClientId();
+	  log.info("clientId = " + clientId2);
+
+	  boolean isConnected = client.isConnected();
+	  log.info("isConnected = " + isConnected);
+
+	  String id = client.getServerURI();
+	  log.info("ServerURI = " + id);
+
+	  log.info("Disconnecting...");
+	  client.disconnect();
+
+	  log.info("Re-Connecting...");
+	  client.connect();
+
+	  log.info("Disconnecting...");
+	  client.disconnect();
+	  
+	  int after_count = Thread.activeCount();
+	  Thread[] tarray = new Thread[after_count];
+	  after_count = Thread.enumerate(tarray);
+	  for (int i = 0; i < after_count; ++i) {
+	    log.info(i + " " + tarray[i].getName());
+	  }
+	  Assert.assertEquals(before_thread_count, after_count);
+	}
+	catch (MqttException exception) {
+	  log.log(Level.SEVERE, "caught exception:", exception);
+	  Assert.fail("Unexpected exception: " + exception);
+	}
+	finally {
+	  if (client != null) {
+	     log.info("Close...");
+	     client.close();
+	   }
+	}
+    int after_count = Thread.activeCount();
+    Thread[] tarray = new Thread[after_count];
+    after_count = Thread.enumerate(tarray);
+    for (int i = 0; i < after_count; ++i) {
+    	  log.info(i + " " + tarray[i].getName());
+    }
+    Assert.assertEquals(before_thread_count, after_count);
+  }
+  
+  
+  @Test
+  public void test402a() throws Exception {
+    String methodName = Utility.getMethodName();
+    LoggingUtilities.banner(log, cclass, methodName);
+
+    IMqttClient client = null;
+    int before_thread_count = Thread.activeCount();
+    final int pool_size = 10;
+    try {
+      String clientId = methodName;
+      client = new MqttClient(serverURI.toString(), clientId, null, Executors.newScheduledThreadPool(pool_size));
+
+      log.info("Connecting...(serverURI:" + serverURI + ", ClientId:" + clientId);
+      client.connect();
+
+	  String clientId2 = client.getClientId();
+	  log.info("clientId = " + clientId2);
+
+	  boolean isConnected = client.isConnected();
+	  log.info("isConnected = " + isConnected);
+
+	  String id = client.getServerURI();
+	  log.info("ServerURI = " + id);
+
+	  log.info("Disconnecting...");
+	  client.disconnect();
+
+	  log.info("Re-Connecting...");
+	  client.connect();
+
+	  log.info("Disconnecting...");
+	  client.disconnect();
+	}
+	catch (MqttException exception) {
+	  log.log(Level.SEVERE, "caught exception:", exception);
+	  Assert.fail("Unexpected exception: " + exception);
+	}
+	finally {
+	  if (client != null) {
+	     log.info("Close...");
+	     client.close();
+	   }
+	}
+    int after_count = Thread.activeCount();
+    Thread[] tarray = new Thread[after_count];
+    after_count = Thread.enumerate(tarray);
+    for (int i = 0; i < after_count; ++i) {
+      log.info(i + " " + tarray[i].getName());
+    }
+    Assert.assertEquals(before_thread_count, after_count - pool_size);
+  }
 
 
   // -------------------------------------------------------------
diff --git a/org.eclipse.paho.client.mqttv3/src/main/java-templates/org/eclipse/paho/client/mqttv3/internal/ClientComms.java b/org.eclipse.paho.client.mqttv3/src/main/java-templates/org/eclipse/paho/client/mqttv3/internal/ClientComms.java
@@ -242,7 +242,8 @@ public void close(boolean force) throws MqttException {
 				}
 
 				conState = CLOSED;
-				shutdownExecutorService();
+				// Don't shut down an externally supplied executor service 
+				//shutdownExecutorService();
 				// ShutdownConnection has already cleaned most things
 				clientState.close();
 				clientState = null;
@@ -688,7 +689,11 @@ private class ConnectBG implements Runnable {
 		}
 
 		void start() {
-			executorService.execute(this);
+			if (executorService == null) {
+				new Thread(this).start();
+			} else {
+				executorService.execute(this);
+			}
 		}
 
 		public void run() {
@@ -753,7 +758,11 @@ private class DisconnectBG implements Runnable {
 
 		void start() {
 			threadName = "MQTT Disc: "+getClient().getClientId();
-			executorService.execute(this);
+			if (executorService == null) {
+				new Thread(this).start();
+			} else {
+				executorService.execute(this);
+			}
 		}
 
 		public void run() {
@@ -865,7 +874,11 @@ public void notifyConnect() {
 			log.fine(CLASS_NAME, methodName, "509", null);
 
 			disconnectedMessageBuffer.setPublishCallback(new ReconnectDisconnectedBufferCallback(methodName));
-			executorService.execute(disconnectedMessageBuffer);
+			if (executorService == null) {
+				new Thread(disconnectedMessageBuffer).start();
+			} else {
+				executorService.execute(disconnectedMessageBuffer);
+			}
 		}
 	}
 	
diff --git a/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java b/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java
@@ -421,8 +421,7 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence
 	 * @param pingSender
 	 *            Custom {@link MqttPingSender} implementation.
 	 * @param executorService
-	 *            used for managing threads. If null then a newScheduledThreadPool
-	 *            is used.
+	 *            used for managing threads. If null no executor service is used.
 	 * @throws IllegalArgumentException
 	 *             if the URI does not start with "tcp://", "ssl://" or
 	 *             "local://"
@@ -463,9 +462,6 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence
 		}
 
 		this.executorService = executorService;
-		if (this.executorService == null) {
-			this.executorService = Executors.newScheduledThreadPool(10);
-		}
 
 		// @TRACE 101=<init> ClientID={0} ServerURI={1} PersistenceType={2}
 		log.fine(CLASS_NAME, methodName, "101", new Object[] { clientId, serverURI, persistence });
diff --git a/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/CommsCallback.java b/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/CommsCallback.java
@@ -96,7 +96,11 @@ public void start(String threadName, ExecutorService executorService) {
 
 				running = true;
 				quiescing = false;
-				callbackFuture = executorService.submit(this);
+				if (executorService == null) {
+					new Thread(this).start();
+				} else {
+					callbackFuture = executorService.submit(this);
+				}
 			}
 		}
 	}
diff --git a/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/CommsReceiver.java b/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/CommsReceiver.java
@@ -74,7 +74,11 @@ public void start(String threadName, ExecutorService executorService) {
 		synchronized (lifecycle) {
 			if (!running) {
 				running = true;
-				receiverFuture = executorService.submit(this);
+				if (executorService == null) {
+					new Thread(this).start();
+				} else {
+					receiverFuture = executorService.submit(this);
+				}
 			}
 		}
 	}

Original file line number	Diff line number	Diff line change
`@@ -96,7 +96,11 @@ public void start(String threadName, ExecutorService executorService) {`
`96`	`96`
`97`	`97`	`running = true;`
`98`	`98`	`quiescing = false;`
`99`		`- callbackFuture = executorService.submit(this);`
	`99`	`+ if (executorService == null) {`
	`100`	`+ new Thread(this).start();`
	`101`	`+ } else {`
	`102`	`+ callbackFuture = executorService.submit(this);`
	`103`	`+ }`
`100`	`104`	`}`
`101`	`105`	`}`
`102`	`106`	`}`
Original file line number	Diff line number	Diff line change
`@@ -74,7 +74,11 @@ public void start(String threadName, ExecutorService executorService) {`
`74`	`74`	`synchronized (lifecycle) {`
`75`	`75`	`if (!running) {`
`76`	`76`	`running = true;`
`77`		`- receiverFuture = executorService.submit(this);`
	`77`	`+ if (executorService == null) {`
	`78`	`+ new Thread(this).start();`
	`79`	`+ } else {`
	`80`	`+ receiverFuture = executorService.submit(this);`
	`81`	`+ }`
`78`	`82`	`}`
`79`	`83`	`}`
`80`	`84`	`}`