adding required chnages

* `ClientManager`, `Mqttv3ClientManager`, `Mqttv5ClientManager`, `AbstractMqttMessageDrivenChannelAdapter` - renamig `isConnection()` to `isConnected()`
* fixing docs `mqtt.adoc` and `whats-new.adoc`
* `ClientManagerBackToBackTests` adding factory interface `MessageDrivenChannelAdapterFactory` to create adapters

Author: Ricore72

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

@@ -74,7 +74,7 @@ public interface ClientManager<T, C> extends SmartLifecycle, MqttComponent<C> {
 	 * @return the managed clients isConnected.
 	 * @since 6.4
 	 */
-	boolean isConnection();
+	boolean isConnected();
 
 	/**
 	 * A contract for a custom callback on {@code connectComplete} event from the client.

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

@@ -200,7 +200,7 @@ public void deliveryComplete(IMqttDeliveryToken token) {
 	}
 
 	@Override
-	public boolean isConnection() {
+	public boolean isConnected() {
 		this.lock.lock();
 		try {
 			IMqttAsyncClient client = getClient();

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

@@ -208,7 +208,7 @@ public void mqttErrorOccurred(MqttException exception) {
 	}
 
 	@Override
-	public boolean isConnection() {
+	public boolean isConnected() {
 		this.lock.lock();
 		try {
 			IMqttAsyncClient client = getClient();

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

@@ -204,7 +204,7 @@ protected void onInit() {
 		super.onInit();
 		if (this.clientManager != null) {
 			this.clientManager.addCallback(this);
-			if (this.clientManager.isConnection()) {
+			if (this.clientManager.isConnected()) {
 				connectComplete(false);
 			}
 		}

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

@@ -16,7 +16,6 @@
 
 package org.springframework.integration.mqtt;
 
-import java.lang.reflect.Constructor;
 import java.nio.charset.StandardCharsets;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -25,6 +24,7 @@
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -35,7 +35,6 @@
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.context.IntegrationFlowContext;
 import org.springframework.integration.endpoint.MessageProducerSupport;
-import org.springframework.integration.mqtt.core.ClientManager;
 import org.springframework.integration.mqtt.core.Mqttv3ClientManager;
 import org.springframework.integration.mqtt.core.Mqttv5ClientManager;
 import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
@@ -84,7 +83,7 @@ void testV3ClientManagerStarted() throws Exception {
 	@Test
 	void testV3ClientManagerRuntime() throws Exception {
 		testSubscribeAndPublishRuntime(Mqttv3ConfigRuntime.class, Mqttv3ConfigRuntime.TOPIC_NAME,
-				Mqttv3ConfigRuntime.subscribedLatch, Mqttv3ConfigRuntime.adapter);
+				Mqttv3ConfigRuntime.subscribedLatch);
 	}
 
 	@Test
@@ -102,7 +101,7 @@ void testV5ClientManagerStarted() throws Exception {
 	@Test
 	void testV5ClientManagerRuntime() throws Exception {
 		testSubscribeAndPublishRuntime(Mqttv5ConfigRuntime.class, Mqttv5ConfigRuntime.TOPIC_NAME,
-				Mqttv5ConfigRuntime.subscribedLatch, Mqttv5ConfigRuntime.adapter);
+				Mqttv5ConfigRuntime.subscribedLatch);
 	}
 
 	private void testSubscribeAndPublish(Class<?> configClass, String topicName, CountDownLatch subscribedLatch)
@@ -131,19 +130,18 @@ private void testSubscribeAndPublish(Class<?> configClass, String topicName, Cou
 		}
 	}
 
-	private void testSubscribeAndPublishRuntime(Class<?> configClass, String topicName, CountDownLatch subscribedLatch, Class<?> adapter)
+	private void testSubscribeAndPublishRuntime(Class<?> configClass, String topicName, CountDownLatch subscribedLatch)
 			throws Exception {
 
 		try (var ctx = new AnnotationConfigApplicationContext(configClass)) {
 			// given
 			var input = ctx.getBean("mqttOutFlow.input", MessageChannel.class);
 			var flowContext = ctx.getBean(IntegrationFlowContext.class);
-			var clientManager = ctx.getBean(ClientManager.class);
+			var factory = ctx.getBean(MessageDrivenChannelAdapterFactory.class);
 			var output = new QueueChannel();
-			Class<?>[] parameterTypes = {ClientManager.class, String[].class};
-			Constructor<?> declaredConstructor = adapter.getConstructor(parameterTypes);
+
 			flowContext.registration(IntegrationFlow
-					.from((MessageProducerSupport) declaredConstructor.newInstance(clientManager, new String[] {topicName}))
+					.from(factory.createMessageDrivenAdapter(ctx))
 					.channel(output)
 					.get()).register();
 			String testPayload = "foo";
@@ -279,14 +277,12 @@ public IntegrationFlow mqttInFlow(Mqttv3ClientManager mqttv3ClientManager) {
 
 	@Configuration
 	@EnableIntegration
-	public static class Mqttv3ConfigRuntime {
+	public static class Mqttv3ConfigRuntime implements MessageDrivenChannelAdapterFactory {
 
 		static final String TOPIC_NAME = "test-topic-v3";
 
 		static final CountDownLatch subscribedLatch = new CountDownLatch(1);
 
-		static final Class<?> adapter = MqttPahoMessageDrivenChannelAdapter.class;
-
 		@EventListener
 		public void onSubscribed(MqttSubscribedEvent e) {
 			subscribedLatch.countDown();
@@ -305,6 +301,11 @@ public IntegrationFlow mqttOutFlow(Mqttv3ClientManager mqttv3ClientManager) {
 			return f -> f.handle(new MqttPahoMessageHandler(mqttv3ClientManager));
 		}
 
+		@Override
+		public MessageProducerSupport createMessageDrivenAdapter(ApplicationContext ctx) {
+			var clientManager = ctx.getBean(Mqttv3ClientManager.class);
+			return new MqttPahoMessageDrivenChannelAdapter(clientManager, TOPIC_NAME);
+		}
 	}
 
 	@Configuration
@@ -414,14 +415,12 @@ public IntegrationFlow mqttInFlow(Mqttv5ClientManager mqttv5ClientManager) {
 
 	@Configuration
 	@EnableIntegration
-	public static class Mqttv5ConfigRuntime {
+	public static class Mqttv5ConfigRuntime implements MessageDrivenChannelAdapterFactory {
 
 		static final String TOPIC_NAME = "test-topic-v5";
 
 		static final CountDownLatch subscribedLatch = new CountDownLatch(1);
 
-		static final Class<?> adapter = Mqttv5PahoMessageDrivenChannelAdapter.class;
-
 		@EventListener
 		public void onSubscribed(MqttSubscribedEvent e) {
 			subscribedLatch.countDown();
@@ -438,6 +437,15 @@ public Mqttv5PahoMessageHandler mqttv5PahoMessageHandler(Mqttv5ClientManager mqt
 			return new Mqttv5PahoMessageHandler(mqttv5ClientManager);
 		}
 
+		@Override
+		public MessageProducerSupport createMessageDrivenAdapter(ApplicationContext ctx) {
+			var clientManager = ctx.getBean(Mqttv5ClientManager.class);
+			return new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, TOPIC_NAME);
+		}
+	}
+
+	interface MessageDrivenChannelAdapterFactory {
+		MessageProducerSupport createMessageDrivenAdapter(ApplicationContext ctx);
 	}
 
 	record ClientV3Disconnector(Mqttv3ClientManager clientManager) {

src/reference/antora/modules/ROOT/pages/mqtt.adoc

@@ -382,9 +382,9 @@ public class MqttJavaApplication {
             .run(args);
     }
 
-   	@Bean
-   	public IntegrationFlow mqttOutboundFlow() {
-   	    return f -> f.handle(new MqttPahoMessageHandler("tcp://host1:1883", "someMqttClient"));
+       @Bean
+       public IntegrationFlow mqttOutboundFlow() {
+           return f -> f.handle(new MqttPahoMessageHandler("tcp://host1:1883", "someMqttClient"));
     }
 
 }
@@ -553,14 +553,14 @@ NOTE: Starting with version 6.4, multiple instances of `MqttPahoMessageDrivenCha
 
 [source,java]
 ----
-private void addAddRuntimeAdapter(IntegrationFlowContext flowContext,Mqttv5ClientManager clientManager,
-								  String topic,MessageChannel channel){
-	flowContext
-		.registration(
-			IntegrationFlow
-				.from(new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, topic))
-				.channel(channel)
-				.get())
-		.register();
+private void addAddRuntimeAdapter(IntegrationFlowContext flowContext, Mqttv5ClientManager clientManager,
+                                  String topic, MessageChannel channel) {
+    flowContext
+        .registration(
+            IntegrationFlow
+                .from(new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, topic))
+                .channel(channel)
+                .get())
+        .register();
 }
 ----

src/reference/antora/modules/ROOT/pages/whats-new.adoc

@@ -68,8 +68,8 @@ See xref:control-bus.adoc[Control Bus] for more information.
 The `DefaultSftpSessionFactory` now exposes a `Consumer<SshClient>` configurer property to further customize an internal `SshClient`.
 See xref:sftp/session-factory.adoc[SFTP Session Factory] for more information.
 
-[[x6.4-mqtt-changes]]
-=== mqtt Changes
+[[x6.4-mqtt-support-changes]]
+=== MQTT Support Changes
 
 Multiple instances of `MqttPahoMessageDrivenChannelAdapter` and `Mqttv5PahoMessageDrivenChannelAdapter` can now be added at runtime using corresponding `ClientManager` through `IntegrationFlowContext`
 See xref:mqtt.adoc[MQTT Support] for more information.