Gh 4014

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

@@ -43,6 +43,12 @@
  */
 public abstract class AbstractMqttClientManager<T, C> implements ClientManager<T, C>, ApplicationEventPublisherAware {
 
+	/**
+	 * The default completion timeout in milliseconds.
+	 * @since 6.0.3
+	 */
+	public static final long DEFAULT_COMPLETION_TIMEOUT = 30_000L;
+
 	protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
 
 	private static final int DEFAULT_MANAGER_PHASE = 0;
@@ -67,6 +73,8 @@ public abstract class AbstractMqttClientManager<T, C> implements ClientManager<T
 
 	private T client;
 
+	private long completionTimeout = DEFAULT_COMPLETION_TIMEOUT;
+
 	protected AbstractMqttClientManager(String clientId) {
 		Assert.notNull(clientId, "'clientId' is required");
 		this.clientId = clientId;
@@ -154,6 +162,21 @@ public String getBeanName() {
 		return this.beanName;
 	}
 
+	/**
+	 * Set the completion timeout for operations. Not settable using the namespace.
+	 * Default {@value #DEFAULT_COMPLETION_TIMEOUT} milliseconds.
+	 * @param completionTimeout The timeout.
+	 * @since 6.0.3
+	 * @author Jiří Patera
+	 */
+	public void setCompletionTimeout(long completionTimeout) {
+		this.completionTimeout = completionTimeout;
+	}
+
+	protected long getCompletionTimeout() {
+		return this.completionTimeout;
+	}
+
 	/**
 	 * The phase of component auto-start in {@link SmartLifecycle}.
 	 * If the custom one is required, note that for the correct behavior it should be less than phase of

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

@@ -139,7 +139,7 @@ public synchronized void stop() {
 			return;
 		}
 		try {
-			client.disconnectForcibly(getDisconnectCompletionTimeout());
+			client.disconnectForcibly(getCompletionTimeout());
 		}
 		catch (MqttException e) {
 			logger.error("Could not disconnect from the client", e);

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

@@ -141,7 +141,7 @@ public synchronized void stop() {
 		}
 
 		try {
-			client.disconnectForcibly(getDisconnectCompletionTimeout());
+			client.disconnectForcibly(getCompletionTimeout());
 		}
 		catch (MqttException e) {
 			logger.error("Could not disconnect from the client", e);

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

@@ -64,6 +64,13 @@ public class MqttPahoMessageDrivenChannelAdapter
 		extends AbstractMqttMessageDrivenChannelAdapter<IMqttAsyncClient, MqttConnectOptions>
 		implements MqttCallbackExtended, MqttPahoComponent {
 
+	/**
+	 * The default disconnect completion timeout in milliseconds.
+	 * @deprecated As of release 6.0.3, replaced by {@link AbstractMqttMessageDrivenChannelAdapter#DEFAULT_COMPLETION_TIMEOUT}
+	 */
+	@Deprecated(since = "6.0.3")
+	public static final long DISCONNECT_COMPLETION_TIMEOUT = DEFAULT_COMPLETION_TIMEOUT;
+
 	private final MqttPahoClientFactory clientFactory;
 
 	private volatile IMqttAsyncClient client;
@@ -131,6 +138,18 @@ public MqttPahoMessageDrivenChannelAdapter(ClientManager<IMqttAsyncClient, MqttC
 		this.clientFactory = factory;
 	}
 
+	/**
+	 * Set the completion timeout when disconnecting. Not settable using the namespace.
+	 * Default {@value #DISCONNECT_COMPLETION_TIMEOUT} milliseconds.
+	 * @param completionTimeout The timeout.
+	 * @since 5.1.10
+	 * @deprecated As of release 6.0.3, replaced by {@link #setCompletionTimeout(long)}
+	 */
+	@Deprecated(since = "6.0.3")
+	public synchronized void setDisconnectCompletionTimeout(long completionTimeout) {
+		setCompletionTimeout(completionTimeout);
+	}
+
 	@Override
 	public MqttConnectOptions getConnectionInfo() {
 		MqttConnectOptions options = this.clientFactory.getConnectionOptions();
@@ -226,7 +245,7 @@ protected synchronized void doStop() {
 		}
 
 		try {
-			this.client.disconnectForcibly(getDisconnectCompletionTimeout());
+			this.client.disconnectForcibly(getCompletionTimeout());
 		}
 		catch (MqttException ex) {
 			logger.error(ex, "Exception while disconnecting");

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java

@@ -512,7 +512,7 @@ public void testDifferentQos() throws Exception {
 
 		new DirectFieldAccessor(adapter).setPropertyValue("running", Boolean.TRUE);
 		adapter.stop();
-		verify(client).disconnectForcibly(5_000L);
+		verify(client).disconnectForcibly(30_000L);
 	}
 
 	private MqttPahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttAsyncClient client, Boolean cleanSession)