Use ScheduledExecutor for the PingSender (#364)

The use of Timer, which handles its own thread, was causing exceptions
when used in a JEE context when the heart beat was missed: the
reconnect, which was being executed from the heart beat’s Timer thread,
was trying to submit to the ManagedExecutorService, which is not
allowed.

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

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

Author: david-katz

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java

@@ -27,8 +27,8 @@
 import java.util.Properties;
 import java.util.Timer;
 import java.util.TimerTask;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 
 import javax.net.SocketFactory;
 import javax.net.ssl.SSLSocketFactory;
@@ -112,7 +112,7 @@ public class MqttAsyncClient implements IMqttAsyncClient {
 
 
 
-	private ExecutorService executorService;
+	private ScheduledExecutorService executorService;
 
 	/**
 	 * Create an MqttAsyncClient that is used to communicate with an MQTT server.
@@ -366,8 +366,8 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence
 	 * @throws IllegalArgumentException if the clientId is null or is greater than 65535 characters in length
 	 * @throws MqttException if any other problem was encountered
 	 */
-	public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence persistence, MqttPingSender pingSender, ExecutorService executorService) throws MqttException {
-		final String methodName = "MqttAsyncClient";
+	public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence persistence, MqttPingSender pingSender, ScheduledExecutorService executorService) throws MqttException {
+        	final String methodName = "MqttAsyncClient";
 
 		log.setResourceName(clientId);
 
@@ -397,7 +397,7 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence
 
 		this.executorService = executorService;
 		if (this.executorService == null) {
-			this.executorService = Executors.newFixedThreadPool(10);
+			this.executorService = Executors.newScheduledThreadPool(10);
 		}
 
 		// @TRACE 101=<init> ClientID={0} ServerURI={1} PersistenceType={2}

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttClient.java

@@ -19,7 +19,7 @@
 package org.eclipse.paho.client.mqttv3;
 
 import java.util.Properties;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
 
 import javax.net.SocketFactory;
 
@@ -315,8 +315,8 @@ public MqttClient(String serverURI, String clientId, MqttClientPersistence persi
 	 * @throws IllegalArgumentException if the clientId is null or is greater than 65535 characters in length
 	 * @throws MqttException if any other problem was encountered
 	 */
-	public MqttClient(String serverURI, String clientId, MqttClientPersistence persistence, ExecutorService executorService) throws MqttException {
-		aClient = new MqttAsyncClient(serverURI, clientId, persistence, new TimerPingSender(), executorService);
+	public MqttClient(String serverURI, String clientId, MqttClientPersistence persistence, ScheduledExecutorService executorService) throws MqttException {
+		aClient = new MqttAsyncClient(serverURI, clientId, persistence, new ScheduledExecutorPingSender(executorService), executorService);
 	}
 
 	/*

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/ScheduledExecutorPingSender.java

@@ -0,0 +1,92 @@
+/*******************************************************************************
+ * Copyright (c) 2014 IBM Corp.
+ * Copyright (c) 2017 BMW Car IT GmbH.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Eclipse Distribution License v1.0 which accompany this distribution.
+ *
+ * The Eclipse Public License is available at
+ *    http://www.eclipse.org/legal/epl-v10.html
+ * and the Eclipse Distribution License is available at
+ *   http://www.eclipse.org/org/documents/edl-v10.php.
+ */
+
+package org.eclipse.paho.client.mqttv3;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+import org.eclipse.paho.client.mqttv3.internal.ClientComms;
+import org.eclipse.paho.client.mqttv3.logging.Logger;
+import org.eclipse.paho.client.mqttv3.logging.LoggerFactory;
+
+/**
+ * Default ping sender implementation
+ *
+ * <p>This class implements the {@link IMqttPingSender} pinger interface
+ * allowing applications to send ping packet to server every keep alive interval.
+ * </p>
+ *
+ * @see MqttPingSender
+ */
+public class ScheduledExecutorPingSender implements MqttPingSender {
+	private static final String CLASS_NAME = ScheduledExecutorPingSender.class.getName();
+	private static final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
+
+	private ClientComms comms;
+	private ScheduledExecutorService executorService;
+	private ScheduledFuture scheduledFuture;
+	private String clientid;
+
+	public ScheduledExecutorPingSender(ScheduledExecutorService executorService) {
+		if (executorService == null) {
+			throw new IllegalArgumentException("ExecutorService cannot be null.");
+		}
+		this.executorService = executorService;
+	}
+
+	public void init(ClientComms comms) {
+		if (comms == null) {
+			throw new IllegalArgumentException("ClientComms cannot be null.");
+		}
+		this.comms = comms;
+		clientid = comms.getClient().getClientId();
+	}
+
+	public void start() {
+		final String methodName = "start";
+
+		//@Trace 659=start timer for client:{0}
+		log.fine(CLASS_NAME, methodName, "659", new Object[]{ clientid });
+		//Check ping after first keep alive interval.
+		schedule(comms.getKeepAlive());
+	}
+
+	public void stop() {
+		final String methodName = "stop";
+		//@Trace 661=stop
+		log.fine(CLASS_NAME, methodName, "661", null);
+		if (scheduledFuture != null) {
+			scheduledFuture.cancel(true);
+		}
+	}
+
+	public void schedule(long delayInMilliseconds) {
+		scheduledFuture = executorService.schedule(new PingRunnable(), delayInMilliseconds, TimeUnit.MILLISECONDS);
+	}
+
+	private class PingRunnable implements Runnable {
+		private static final String methodName = "PingTask.run";
+
+		public void run() {
+			String originalThreadName = Thread.currentThread().getName();
+			Thread.currentThread().setName("MQTT Ping: " + clientid);
+			//@Trace 660=Check schedule at {0}
+			log.fine(CLASS_NAME, methodName, "660", new Object[]{ new Long(System.currentTimeMillis()) });
+			comms.checkForActivity();
+			Thread.currentThread().setName(originalThreadName);
+		}
+	}
+}