GH-3685: Share MQTT connection across components

Fixes #3685

Add a couple of unit/integration tests to cover client manager usage.
Several small code improvements after the code review:
* Improve client manager usage via providing several mutual exclusive
constructors, whether the users provides `url` or `connectionOptions`
or `clientFactory` for v3.
* Move the logger to `AbstractMqttClientManager`
* Do not inject TaskScheduler in constructor for v3 client manager
but use lazy init via `BeanFactory` and `IntegrationContextUtils`
* Other smaller code readability improvements

Author: oxcafedead

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/AbstractMqttClientManager.java

@@ -16,20 +16,24 @@
 
 package org.springframework.integration.mqtt.core;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.util.Assert;
 
 public abstract class AbstractMqttClientManager<T> implements ClientManager<T> {
 
+	protected final Log logger = LogFactory.getLog(this.getClass());
+
 	private boolean manualAcks;
 
 	private String url;
 
-	private String clientId;
+	private final String clientId;
 
-	AbstractMqttClientManager(String url, String clientId) {
+	AbstractMqttClientManager(String clientId) {
 		Assert.notNull(clientId, "'clientId' is required");
 		this.clientId = clientId;
-		this.url = url;
 	}
 
 	@Override
@@ -53,7 +57,4 @@ public String getClientId() {
 		return this.clientId;
 	}
 
-	public void setClientId(String clientId) {
-		this.clientId = clientId;
-	}
 }

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv3ClientManager.java

@@ -19,55 +19,76 @@
 import java.time.Instant;
 import java.util.concurrent.ScheduledFuture;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
 import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.scheduling.TaskScheduler;
 import org.springframework.util.Assert;
 
-public class Mqttv3ClientManager extends AbstractMqttClientManager<IMqttAsyncClient> implements MqttCallback {
+public class Mqttv3ClientManager extends AbstractMqttClientManager<IMqttAsyncClient>
+		implements MqttCallback, InitializingBean, BeanFactoryAware {
 
 	/**
 	 * The default reconnect timeout in millis.
 	 */
 	private static final long DEFAULT_RECOVERY_INTERVAL = 10_000;
 
-	private final Log logger = LogFactory.getLog(this.getClass());
-
 	private final MqttPahoClientFactory clientFactory;
 
-	private final TaskScheduler taskScheduler;
+	private BeanFactory beanFactory;
+
+	private TaskScheduler taskScheduler;
 
 	private volatile ScheduledFuture<?> scheduledReconnect;
 
 	private volatile IMqttAsyncClient client;
 
 	private long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
 
-	public Mqttv3ClientManager(MqttPahoClientFactory clientFactory, TaskScheduler taskScheduler, String url,
-			String clientId) {
-
-		super(url, clientId);
-		Assert.notNull(clientId, "'clientFactory' is required");
-		Assert.notNull(clientId, "'taskScheduler' is required");
-		if (url == null) {
-			Assert.notEmpty(clientFactory.getConnectionOptions().getServerURIs(), "'serverURIs' must be provided in the 'MqttConnectionOptions'");
-		}
+	public Mqttv3ClientManager(MqttPahoClientFactory clientFactory, String clientId) {
+		super(clientId);
+		Assert.notNull(clientFactory, "'clientFactory' is required");
 		this.clientFactory = clientFactory;
-		this.taskScheduler = taskScheduler;
+		String[] serverURIs = clientFactory.getConnectionOptions().getServerURIs();
+		Assert.notEmpty(serverURIs, "'serverURIs' must be provided in the 'MqttConnectionOptions'");
+		setUrl(serverURIs[0]);
+	}
+
+	public Mqttv3ClientManager(String url, String clientId) {
+		super(clientId);
+		Assert.notNull(url, "'url' is required");
+		setUrl(url);
+		MqttConnectOptions connectOptions = new MqttConnectOptions();
+		connectOptions.setServerURIs(new String[]{ url });
+		DefaultMqttPahoClientFactory defaultFactory = new DefaultMqttPahoClientFactory();
+		defaultFactory.setConnectionOptions(connectOptions);
+		this.clientFactory = defaultFactory;
 	}
 
 	@Override
 	public IMqttAsyncClient getClient() {
 		return this.client;
 	}
 
+	@Override
+	public void afterPropertiesSet() {
+		this.taskScheduler = IntegrationContextUtils.getTaskScheduler(this.beanFactory);
+	}
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) {
+		Assert.notNull(beanFactory, "'beanFactory' must not be null");
+		this.beanFactory = beanFactory;
+	}
+
 	@Override
 	public synchronized void start() {
 		if (this.client == null) {
@@ -84,7 +105,7 @@ public synchronized void start() {
 			connect();
 		}
 		catch (MqttException e) {
-			this.logger.error("could not start client manager, scheduling reconnect, client_id=" +
+			logger.error("could not start client manager, scheduling reconnect, client_id=" +
 					this.client.getClientId(), e);
 			scheduleReconnect();
 		}
@@ -99,14 +120,14 @@ public synchronized void stop() {
 			this.client.disconnectForcibly(this.clientFactory.getConnectionOptions().getConnectionTimeout());
 		}
 		catch (MqttException e) {
-			this.logger.error("could not disconnect from the client", e);
+			logger.error("could not disconnect from the client", e);
 		}
 		finally {
 			try {
 				this.client.close();
 			}
 			catch (MqttException e) {
-				this.logger.error("could not close the client", e);
+				logger.error("could not close the client", e);
 			}
 			this.client = null;
 		}
@@ -119,9 +140,9 @@ public synchronized boolean isRunning() {
 
 	@Override
 	public synchronized void connectionLost(Throwable cause) {
-		this.logger.error("connection lost, scheduling reconnect, client_id=" + this.client.getClientId(),
+		logger.error("connection lost, scheduling reconnect, client_id=" + this.client.getClientId(),
 				cause);
-		scheduleReconnect(); // todo: do we need to resubscribe if con lost?
+		scheduleReconnect();
 	}
 
 	@Override
@@ -161,7 +182,7 @@ private synchronized void scheduleReconnect() {
 				this.scheduledReconnect = null;
 			}
 			catch (MqttException e) {
-				this.logger.error("could not reconnect", e);
+				logger.error("could not reconnect", e);
 				scheduleReconnect();
 			}
 		}, Instant.now().plusMillis(getRecoveryInterval()));

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv5ClientManager.java

@@ -16,8 +16,6 @@
 
 package org.springframework.integration.mqtt.core;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
 import org.eclipse.paho.mqttv5.client.IMqttAsyncClient;
 import org.eclipse.paho.mqttv5.client.IMqttToken;
 import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
@@ -32,19 +30,29 @@
 
 public class Mqttv5ClientManager extends AbstractMqttClientManager<IMqttAsyncClient> implements MqttCallback {
 
-	private final Log logger = LogFactory.getLog(this.getClass());
-
 	private final MqttConnectionOptions connectionOptions;
 
 	private volatile IMqttAsyncClient client;
 
-	public Mqttv5ClientManager(MqttConnectionOptions connectionOptions, String url, String clientId) {
-		super(url, clientId);
+	public Mqttv5ClientManager(String url, String clientId) {
+		super(clientId);
+		Assert.notNull(url, "'url' is required");
+		setUrl(url);
+		this.connectionOptions = new MqttConnectionOptions();
+		this.connectionOptions.setServerURIs(new String[]{ url });
+		this.connectionOptions.setAutomaticReconnect(true);
+	}
+
+	public Mqttv5ClientManager(MqttConnectionOptions connectionOptions, String clientId) {
+		super(clientId);
 		Assert.notNull(connectionOptions, "'connectionOptions' is required");
-		if (url == null) {
-			Assert.notEmpty(connectionOptions.getServerURIs(), "'serverURIs' must be provided in the 'MqttConnectionOptions'");
-		}
 		this.connectionOptions = connectionOptions;
+		if (!this.connectionOptions.isAutomaticReconnect()) {
+			logger.warn("It is recommended to set 'automaticReconnect' MQTT connection option. " +
+					"Otherwise connection check and reconnect should be done manually.");
+		}
+		Assert.notEmpty(connectionOptions.getServerURIs(), "'serverURIs' must be provided in the 'MqttConnectionOptions'");
+		setUrl(connectionOptions.getServerURIs()[0]);
 	}
 
 	@Override
@@ -69,7 +77,7 @@ public synchronized void start() {
 					.waitForCompletion(this.connectionOptions.getConnectionTimeout());
 		}
 		catch (MqttException e) {
-			this.logger.error("could not start client manager, client_id=" + this.client.getClientId(), e);
+			logger.error("could not start client manager, client_id=" + this.client.getClientId(), e);
 		}
 	}
 
@@ -83,14 +91,14 @@ public synchronized void stop() {
 			this.client.disconnectForcibly(this.connectionOptions.getConnectionTimeout());
 		}
 		catch (MqttException e) {
-			this.logger.error("could not disconnect from the client", e);
+			logger.error("could not disconnect from the client", e);
 		}
 		finally {
 			try {
 				this.client.close();
 			}
 			catch (MqttException e) {
-				this.logger.error("could not close the client", e);
+				logger.error("could not close the client", e);
 			}
 			this.client = null;
 		}
@@ -113,8 +121,8 @@ public void deliveryComplete(IMqttToken token) {
 
 	@Override
 	public void connectComplete(boolean reconnect, String serverURI) {
-		if (this.logger.isInfoEnabled()) {
-			this.logger.info("MQTT connect complete to " + serverURI);
+		if (logger.isInfoEnabled()) {
+			logger.info("MQTT connect complete to " + serverURI);
 		}
 		// probably makes sense to use custom callbacks in the future
 	}
@@ -126,14 +134,14 @@ public void authPacketArrived(int reasonCode, MqttProperties properties) {
 
 	@Override
 	public void disconnected(MqttDisconnectResponse disconnectResponse) {
-		if (this.logger.isInfoEnabled()) {
-			this.logger.info("MQTT disconnected" + disconnectResponse);
+		if (logger.isInfoEnabled()) {
+			logger.info("MQTT disconnected" + disconnectResponse);
 		}
 	}
 
 	@Override
 	public void mqttErrorOccurred(MqttException exception) {
-		this.logger.error("MQTT error occurred", exception);
+		logger.error("MQTT error occurred", exception);
 	}
 
 }

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

@@ -331,11 +331,9 @@ public void deliveryComplete(IMqttToken token) {
 
 	@Override
 	public void connectComplete(boolean reconnect, String serverURI) {
-		if (reconnect) {
-			return;
+		if (!reconnect) {
+			subscribeToAll();
 		}
-
-		subscribeToAll();
 	}
 
 	@Override