adding runtime ClientManger tests

* `testV3ClientManagerRuntime` and `testV5ClientManagerRuntime` adding MessageDrivenAdaptes at runtime using `IntegrationFlowContext`

Author: Ricore72

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

@@ -16,7 +16,9 @@
 
 package org.springframework.integration.mqtt;
 
+import java.lang.reflect.Constructor;
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -29,8 +31,12 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.event.EventListener;
 import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 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;
@@ -76,6 +82,12 @@ void testV3ClientManagerStarted() throws Exception {
 				Mqttv3ConfigWithStartedManager.subscribedLatch);
 	}
 
+	@Test
+	void testV3ClientManagerRuntime() throws Exception{
+		testSubscribeAndPublishRuntime(Mqttv3ConfigRuntime.class, Mqttv3ConfigRuntime.TOPIC_NAME,
+				Mqttv3ConfigRuntime.subscribedLatch, Mqttv3ConfigRuntime.adapter);
+	}
+
 	@Test
 	void testV5ClientManagerReconnect() throws Exception {
 		testSubscribeAndPublish(Mqttv5ConfigWithDisconnect.class, Mqttv5ConfigWithDisconnect.TOPIC_NAME,
@@ -88,6 +100,12 @@ void testV5ClientManagerStarted() throws Exception {
 				Mqttv5ConfigWithStartedManager.subscribedLatch);
 	}
 
+	@Test
+	void testV5ClientManagerRuntime() throws Exception{
+		testSubscribeAndPublishRuntime(Mqttv5ConfigRuntime.class, Mqttv5ConfigRuntime.TOPIC_NAME,
+				Mqttv5ConfigRuntime.subscribedLatch, Mqttv5ConfigRuntime.adapter);
+	}
+
 	private void testSubscribeAndPublish(Class<?> configClass, String topicName, CountDownLatch subscribedLatch)
 			throws Exception {
 
@@ -114,6 +132,40 @@ private void testSubscribeAndPublish(Class<?> configClass, String topicName, Cou
 		}
 	}
 
+	private void testSubscribeAndPublishRuntime(Class<?> configClass, String topicName, CountDownLatch subscribedLatch, Class<?> adapter)
+			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 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}))
+					.channel(output)
+					.get()).register();
+			String testPayload = "foo";
+			assertThat(subscribedLatch.await(20, TimeUnit.SECONDS)).isTrue();
+
+			// when
+			input.send(MessageBuilder.withPayload(testPayload).setHeader(MqttHeaders.TOPIC, topicName).build());
+			Message<?> receive = output.receive(20_000);
+
+			// then
+			assertThat(receive).isNotNull();
+			Object payload = receive.getPayload();
+			if (payload instanceof String sp) {
+				assertThat(sp).isEqualTo(testPayload);
+			}
+			else {
+				assertThat(payload).isEqualTo(testPayload.getBytes(StandardCharsets.UTF_8));
+			}
+		}
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class Mqttv3Config {
@@ -208,6 +260,7 @@ public Mqttv3ClientManager mqttv3ClientManager() {
 			connectionOptions.setServerURIs(new String[] {MosquittoContainerTest.mqttUrl()});
 			connectionOptions.setAutomaticReconnect(true);
 			Mqttv3ClientManager manager = new Mqttv3ClientManager(connectionOptions, "client-manager-client-id-v3");
+			manager.start();
 			return manager;
 		}
 
@@ -223,6 +276,35 @@ public IntegrationFlow mqttInFlow(Mqttv3ClientManager mqttv3ClientManager) {
 					.get();
 		}
 
+	}
+	@Configuration
+	@EnableIntegration
+	public static class Mqttv3ConfigRuntime {
+
+		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();
+		}
+
+		@Bean
+		public Mqttv3ClientManager mqttv3ClientManager() {
+			MqttConnectOptions connectionOptions = new MqttConnectOptions();
+			connectionOptions.setServerURIs(new String[] {MosquittoContainerTest.mqttUrl()});
+			connectionOptions.setAutomaticReconnect(true);
+            return new Mqttv3ClientManager(connectionOptions, "client-manager-client-id-v3");
+		}
+
+		@Bean
+		public IntegrationFlow mqttOutFlow(Mqttv3ClientManager mqttv3ClientManager) {
+			return f -> f.handle(new MqttPahoMessageHandler(mqttv3ClientManager));
+		}
+
 	}
 
 	@Configuration
@@ -328,6 +410,33 @@ public IntegrationFlow mqttInFlow(Mqttv5ClientManager mqttv5ClientManager) {
 					.get();
 		}
 
+	}
+	@Configuration
+	@EnableIntegration
+	public static class Mqttv5ConfigRuntime {
+
+		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();
+		}
+
+		@Bean
+		public Mqttv5ClientManager mqttv5ClientManager() {
+			return new Mqttv5ClientManager(MosquittoContainerTest.mqttUrl(), "client-manager-client-id-v5");
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "mqttOutFlow.input")
+		public Mqttv5PahoMessageHandler mqttv5PahoMessageHandler(Mqttv5ClientManager mqttv5ClientManager) {
+			return new Mqttv5PahoMessageHandler(mqttv5ClientManager);
+		}
+
 	}
 
 	record ClientV3Disconnector(Mqttv3ClientManager clientManager) {