Rework all the MQTT tests to JUnit 5

* Use `MosquittoContainerTest` for all the MQTT integration tests
* Remove MQTT `BrokerRunning` JUnit 4 rule
* Mark `BackToBackAdapterTests` with `@LongRunningTest` as it is really too slow

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests-context.xml

@@ -1,11 +1,13 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beans xmlns="http://www.springframework.org/schema/beans"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xmlns:int="http://www.springframework.org/schema/integration"
-	xmlns:int-mqtt="http://www.springframework.org/schema/integration/mqtt"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:int="http://www.springframework.org/schema/integration"
+	   xmlns:int-mqtt="http://www.springframework.org/schema/integration/mqtt"
+	   xmlns:util="http://www.springframework.org/schema/util"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
 		http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
-		http://www.springframework.org/schema/integration/mqtt https://www.springframework.org/schema/integration/mqtt/spring-integration-mqtt.xsd">
+		http://www.springframework.org/schema/integration/mqtt https://www.springframework.org/schema/integration/mqtt/spring-integration-mqtt.xsd
+		http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd">
 
 	<int-mqtt:outbound-channel-adapter id="out" client-id="multiOut"
 		client-factory="multiUriClientFactory"
@@ -19,10 +21,14 @@
 		<int:queue />
 	</int:channel>
 
+	<bean id="mqtt_url" class="java.lang.String">
+		<constructor-arg value="#{T(org.springframework.integration.mqtt.MosquittoContainerTest).mqttUrl()}"/>
+	</bean>
+
 	<bean id="multiUriClientFactory" class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
 		<property name="connectionOptions">
 			<bean class="org.eclipse.paho.client.mqttv3.MqttConnectOptions">
-				<property name="serverURIs" value="tcp://localhost:1883,tcp://localhost:1883"/>
+				<property name="serverURIs" value="#{mqtt_url},#{mqtt_url}"/>
 			</bean>
 		</property>
 	</bean>

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,20 +17,21 @@
 package org.springframework.integration.mqtt;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.Mockito.mock;
 
-import java.util.Arrays;
+import java.io.File;
 import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
 import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -50,6 +51,7 @@ import org.springframework.integration.mqtt.support.MqttHeaders;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper;
 import org.springframework.integration.support.json.JacksonJsonUtils;
+import org.springframework.integration.test.condition.LongRunningTest;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessagingException;
@@ -57,7 +59,7 @@ import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Gary Russell
@@ -66,15 +68,15 @@ import org.springframework.test.context.junit4.SpringRunner;
  * @since 4.0
  *
  */
-@RunWith(SpringRunner.class)
+@LongRunningTest
+@SpringJUnitConfig
 @DirtiesContext
-public class BackToBackAdapterTests {
+public class BackToBackAdapterTests implements MosquittoContainerTest {
 
-	@ClassRule
-	public static final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883);
+	@TempDir
+	static File folder;
 
-	@ClassRule
-	public static final TemporaryFolder folder = new TemporaryFolder();
+	static ThreadPoolTaskScheduler taskScheduler;
 
 	@Autowired
 	private MessageChannel out;
@@ -85,31 +87,40 @@ public class BackToBackAdapterTests {
 	@Autowired
 	private EventsListener listener;
 
+	@BeforeAll
+	static void setup() {
+		taskScheduler = new ThreadPoolTaskScheduler();
+		taskScheduler.initialize();
+	}
+
+	@AfterAll
+	static void teardown() {
+		taskScheduler.destroy();
+	}
+
 	@Test
 	public void testSingleTopic() {
-		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
+		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler(MosquittoContainerTest.mqttUrl(), "si-test-out");
 		adapter.setDefaultTopic("mqtt-foo");
 		adapter.setBeanFactory(mock(BeanFactory.class));
 		adapter.afterPropertiesSet();
 		adapter.start();
-		MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883",
+		MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter(MosquittoContainerTest.mqttUrl(),
 				"si-test-in", "mqtt-foo");
 		QueueChannel outputChannel = new QueueChannel();
 		inbound.setOutputChannel(outputChannel);
-		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
-		taskScheduler.initialize();
 		inbound.setTaskScheduler(taskScheduler);
 		inbound.setBeanFactory(mock(BeanFactory.class));
 		inbound.afterPropertiesSet();
 		inbound.start();
-		adapter.handleMessage(new GenericMessage<String>("foo"));
+		adapter.handleMessage(new GenericMessage<>("foo"));
 		Message<?> out = outputChannel.receive(20000);
 		assertThat(out).isNotNull();
 		adapter.stop();
 		inbound.stop();
 		assertThat(out.getPayload()).isEqualTo("foo");
 		assertThat(out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)).isEqualTo("mqtt-foo");
-		assertThat(adapter.getConnectionInfo().getServerURIs()[0]).isEqualTo("tcp://localhost:1883");
+		assertThat(adapter.getConnectionInfo().getServerURIs()[0]).isEqualTo(MosquittoContainerTest.mqttUrl());
 	}
 
 	@Test
@@ -123,7 +134,7 @@ public class BackToBackAdapterTests {
 	}
 
 	private void testJsonCommon(String... trusted) {
-		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
+		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler(MosquittoContainerTest.mqttUrl(), "si-test-out");
 		adapter.setDefaultTopic("mqtt-foo");
 		adapter.setBeanFactory(mock(BeanFactory.class));
 		EmbeddedJsonHeadersMessageMapper mapper = new EmbeddedJsonHeadersMessageMapper(
@@ -133,18 +144,16 @@ public class BackToBackAdapterTests {
 		adapter.setConverter(converter);
 		adapter.afterPropertiesSet();
 		adapter.start();
-		MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883",
-				"si-test-in", "mqtt-foo");
+		MqttPahoMessageDrivenChannelAdapter inbound =
+				new MqttPahoMessageDrivenChannelAdapter(MosquittoContainerTest.mqttUrl(), "si-test-in", "mqtt-foo");
 		QueueChannel outputChannel = new QueueChannel();
 		inbound.setOutputChannel(outputChannel);
-		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
-		taskScheduler.initialize();
 		inbound.setTaskScheduler(taskScheduler);
 		inbound.setBeanFactory(mock(BeanFactory.class));
 		inbound.setConverter(converter);
 		inbound.afterPropertiesSet();
 		inbound.start();
-		adapter.handleMessage(new GenericMessage<Foo>(new Foo("bar"), Collections.singletonMap("baz", "qux")));
+		adapter.handleMessage(new GenericMessage<>(new Foo("bar"), Collections.singletonMap("baz", "qux")));
 		Message<?> out = outputChannel.receive(20000);
 		assertThat(out).isNotNull();
 		adapter.stop();
@@ -161,22 +170,21 @@ public class BackToBackAdapterTests {
 
 	@Test
 	public void testAddRemoveTopic() {
-		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
+		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler(MosquittoContainerTest.mqttUrl(), "si-test-out");
 		adapter.setDefaultTopic("mqtt-foo");
 		adapter.setBeanFactory(mock(BeanFactory.class));
 		adapter.afterPropertiesSet();
 		adapter.start();
-		MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in");
+		MqttPahoMessageDrivenChannelAdapter inbound =
+				new MqttPahoMessageDrivenChannelAdapter(MosquittoContainerTest.mqttUrl(), "si-test-in");
 		QueueChannel outputChannel = new QueueChannel();
 		inbound.setOutputChannel(outputChannel);
-		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
-		taskScheduler.initialize();
 		inbound.setTaskScheduler(taskScheduler);
 		inbound.setBeanFactory(mock(BeanFactory.class));
 		inbound.afterPropertiesSet();
 		inbound.start();
 		inbound.addTopic("mqtt-foo");
-		adapter.handleMessage(new GenericMessage<String>("foo"));
+		adapter.handleMessage(new GenericMessage<>("foo"));
 		Message<?> out = outputChannel.receive(20_000);
 		assertThat(out).isNotNull();
 		assertThat(out.getPayload()).isEqualTo("foo");
@@ -194,17 +202,13 @@ public class BackToBackAdapterTests {
 		out = outputChannel.receive(1);
 		assertThat(out).isNull();
 
-		try {
-			inbound.addTopic("mqtt-foo");
-			fail("Expected exception");
-		}
-		catch (MessagingException e) {
-			assertThat(e.getMessage()).isEqualTo("Topic 'mqtt-foo' is already subscribed.");
-		}
+		assertThatExceptionOfType(MessagingException.class)
+				.isThrownBy(() -> inbound.addTopic("mqtt-foo"))
+				.withMessage("Topic 'mqtt-foo' is already subscribed.");
 
 		inbound.addTopic("mqqt-bar", "mqqt-baz");
 		inbound.removeTopic("mqqt-bar", "mqqt-baz");
-		inbound.addTopics(new String[] { "mqqt-bar", "mqqt-baz" }, new int[] { 0, 0 });
+		inbound.addTopics(new String[]{ "mqqt-bar", "mqqt-baz" }, new int[]{ 0, 0 });
 		inbound.removeTopic("mqqt-bar", "mqqt-baz");
 
 		adapter.stop();
@@ -213,22 +217,21 @@ public class BackToBackAdapterTests {
 
 	@Test
 	public void testTwoTopics() {
-		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
+		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler(MosquittoContainerTest.mqttUrl(), "si-test-out");
 		adapter.setDefaultTopic("mqtt-foo");
 		adapter.setBeanFactory(mock(BeanFactory.class));
 		adapter.afterPropertiesSet();
 		adapter.start();
-		MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883",
-				"si-test-in", "mqtt-foo", "mqtt-bar");
+		MqttPahoMessageDrivenChannelAdapter inbound =
+				new MqttPahoMessageDrivenChannelAdapter(MosquittoContainerTest.mqttUrl(),
+						"si-test-in", "mqtt-foo", "mqtt-bar");
 		QueueChannel outputChannel = new QueueChannel();
 		inbound.setOutputChannel(outputChannel);
-		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
-		taskScheduler.initialize();
 		inbound.setTaskScheduler(taskScheduler);
 		inbound.setBeanFactory(mock(BeanFactory.class));
 		inbound.afterPropertiesSet();
 		inbound.start();
-		adapter.handleMessage(new GenericMessage<String>("foo"));
+		adapter.handleMessage(new GenericMessage<>("foo"));
 		Message<?> message = MessageBuilder.withPayload("bar").setHeader(MqttHeaders.TOPIC, "mqtt-bar").build();
 		adapter.handleMessage(message);
 		Message<?> out = outputChannel.receive(20000);
@@ -240,12 +243,13 @@ public class BackToBackAdapterTests {
 		inbound.stop();
 		assertThat(out.getPayload()).isEqualTo("bar");
 		assertThat(out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)).isEqualTo("mqtt-bar");
+
 		adapter.stop();
 	}
 
 	@Test
 	public void testAsync() throws Exception {
-		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
+		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler(MosquittoContainerTest.mqttUrl(), "si-test-out");
 		adapter.setDefaultTopic("mqtt-foo");
 		adapter.setBeanFactory(mock(BeanFactory.class));
 		adapter.setAsync(true);
@@ -255,22 +259,21 @@ public class BackToBackAdapterTests {
 		adapter.afterPropertiesSet();
 		adapter.start();
 		MqttPahoMessageDrivenChannelAdapter inbound =
-				new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in", "mqtt-foo");
+				new MqttPahoMessageDrivenChannelAdapter(MosquittoContainerTest.mqttUrl(), "si-test-in", "mqtt-foo");
 		QueueChannel outputChannel = new QueueChannel();
 		inbound.setOutputChannel(outputChannel);
-		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
-		taskScheduler.initialize();
 		inbound.setTaskScheduler(taskScheduler);
 		inbound.setBeanFactory(mock(BeanFactory.class));
 		inbound.afterPropertiesSet();
 		inbound.start();
-		GenericMessage<String> message = new GenericMessage<String>("foo");
+		GenericMessage<String> message = new GenericMessage<>("foo");
 		adapter.handleMessage(message);
 		verifyEvents(adapter, publisher, message);
 		Message<?> out = outputChannel.receive(20000);
 		assertThat(out).isNotNull();
 		adapter.stop();
 		inbound.stop();
+
 		assertThat(out.getPayload()).isEqualTo("foo");
 		assertThat(out.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)).isEqualTo("mqtt-foo");
 	}
@@ -278,10 +281,10 @@ public class BackToBackAdapterTests {
 	@Test
 	public void testAsyncPersisted() throws Exception {
 		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
-		MqttClientPersistence persistence = new MqttDefaultFilePersistence(folder.getRoot().getAbsolutePath());
+		MqttClientPersistence persistence = new MqttDefaultFilePersistence(folder.getAbsolutePath());
 		factory.setPersistence(persistence);
-		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out",
-				factory);
+		MqttPahoMessageHandler adapter =
+				new MqttPahoMessageHandler(MosquittoContainerTest.mqttUrl(), "si-test-out", factory);
 		adapter.setDefaultTopic("mqtt-foo");
 		adapter.setBeanFactory(mock(BeanFactory.class));
 		adapter.setAsync(true);
@@ -293,16 +296,15 @@ public class BackToBackAdapterTests {
 		adapter.start();
 
 		MqttPahoMessageDrivenChannelAdapter inbound =
-				new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in", "mqtt-foo", "mqtt-bar");
+				new MqttPahoMessageDrivenChannelAdapter(MosquittoContainerTest.mqttUrl(),
+						"si-test-in", "mqtt-foo", "mqtt-bar");
 		QueueChannel outputChannel = new QueueChannel();
 		inbound.setOutputChannel(outputChannel);
-		ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
-		taskScheduler.initialize();
 		inbound.setTaskScheduler(taskScheduler);
 		inbound.setBeanFactory(mock(BeanFactory.class));
 		inbound.afterPropertiesSet();
 		inbound.start();
-		Message<String> message1 = new GenericMessage<String>("foo");
+		Message<String> message1 = new GenericMessage<>("foo");
 		adapter.handleMessage(message1);
 		verifyEvents(adapter, publisher1, message1);
 
@@ -334,7 +336,7 @@ public class BackToBackAdapterTests {
 
 		assertThat(publisher1.delivered.getClientInstance()).isNotEqualTo(clientInstance);
 
-		Message<?> out = null;
+		Message<?> out;
 		for (int i = 0; i < 4; i++) {
 			out = outputChannel.receive(20000);
 			assertThat(out).isNotNull();
@@ -354,6 +356,7 @@ public class BackToBackAdapterTests {
 
 	private void verifyEvents(MqttPahoMessageHandler adapter, EventPublisher publisher1, Message<String> message1)
 			throws InterruptedException {
+
 		assertThat(publisher1.latch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(publisher1.sent).isNotNull();
 		assertThat(publisher1.delivered).isNotNull();
@@ -373,13 +376,13 @@ public class BackToBackAdapterTests {
 
 	@Test
 	public void testMultiURIs() {
-		out.send(new GenericMessage<String>("foo"));
+		out.send(new GenericMessage<>("foo"));
 		Message<?> message = in.receive(20000);
 		assertThat(message).isNotNull();
 		assertThat(message.getPayload()).isEqualTo("foo");
 		MqttPahoComponent source = this.listener.event.getSourceAsType();
-		assertThat(Arrays.toString(source.getConnectionInfo().getServerURIs()))
-				.isEqualTo("[tcp://localhost:1883, tcp://localhost:1883]");
+		assertThat(source.getConnectionInfo().getServerURIs())
+				.contains(MosquittoContainerTest.mqttUrl(), MosquittoContainerTest.mqttUrl());
 	}
 
 	public static class EventsListener implements ApplicationListener<MqttSubscribedEvent> {

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

@@ -1,86 +0,0 @@
-/*
- * Copyright 2002-2019 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      https://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.integration.mqtt;
-
-import static org.assertj.core.api.Assumptions.assumeThat;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.eclipse.paho.client.mqttv3.IMqttClient;
-import org.eclipse.paho.client.mqttv3.MqttException;
-import org.junit.rules.TestWatcher;
-import org.junit.runner.Description;
-import org.junit.runners.model.Statement;
-
-import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
-
-/**
- * @author Gary Russell
- * @author Artem Bilan
- *
- * @since 4.0
- *
- */
-public class BrokerRunning extends TestWatcher {
-
-	private static Log logger = LogFactory.getLog(BrokerRunning.class);
-
-	// Static so that we only test once on failure: speeds up test suite
-	private static Map<Integer, Boolean> brokerOnline = new HashMap<>();
-
-	private final int port;
-
-	private BrokerRunning(int port) {
-		this.port = port;
-		brokerOnline.put(port, true);
-	}
-
-	@Override
-	public Statement apply(Statement base, Description description) {
-		assumeThat(brokerOnline.get(port)).isTrue();
-		String url = "tcp://localhost:" + port;
-		IMqttClient client = null;
-		try {
-			client = new DefaultMqttPahoClientFactory().getClientInstance(url, "junit-" + System.currentTimeMillis());
-			client.connect();
-		}
-		catch (MqttException e) {
-			logger.warn("Tests not running because no broker on " + url + ":", e);
-			assumeThat(e).isNull();
-		}
-		finally {
-			if (client != null) {
-				try {
-					client.disconnect();
-					client.close();
-				}
-				catch (MqttException e) {
-				}
-			}
-		}
-		return super.apply(base, description);
-	}
-
-
-	public static BrokerRunning isRunning(int port) {
-		return new BrokerRunning(port);
-	}
-
-}

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/DownstreamExceptionTests-context.xml

@@ -7,12 +7,16 @@
 		http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
 		http://www.springframework.org/schema/integration/mqtt https://www.springframework.org/schema/integration/mqtt/spring-integration-mqtt.xsd">
 
+	<bean id="mqtt_url" class="java.lang.String">
+		<constructor-arg value="#{T(org.springframework.integration.mqtt.MosquittoContainerTest).mqttUrl()}"/>
+	</bean>
+
 	<int-mqtt:message-driven-channel-adapter id="noErrorChannel"
-		url="tcp://localhost:1883"
+		url="#{mqtt_url}"
 		client-id="fooEx1" channel="foo" topics="mqtt-fooEx1" />
 
 	<int-mqtt:message-driven-channel-adapter id="withErrorChannel"
-		url="tcp://localhost:1883"
+		url="#{mqtt_url}"
 		error-channel="errors"
 		client-id="fooEx2" channel="foo" topics="mqtt-fooEx2" />
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 the original author or authors.
+ * Copyright 2014-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,9 +28,7 @@ import java.util.concurrent.CyclicBarrier;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentMatchers;
 
 import org.springframework.beans.DirectFieldAccessor;
@@ -43,7 +41,7 @@ import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Gary Russell
@@ -52,12 +50,9 @@ import org.springframework.test.context.junit4.SpringRunner;
  * @since 4.0
  *
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
-public class DownstreamExceptionTests {
-
-	@ClassRule
-	public static final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883);
+public class DownstreamExceptionTests implements MosquittoContainerTest {
 
 	@Autowired
 	private Service service;
@@ -84,7 +79,7 @@ public class DownstreamExceptionTests {
 			return null;
 		}).when(logger).error(any(Throwable.class), any(Supplier.class));
 		new DirectFieldAccessor(noErrorChannel).setPropertyValue("logger", logger);
-		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
+		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler(MosquittoContainerTest.mqttUrl(), "si-test-out");
 		adapter.setDefaultTopic("mqtt-fooEx1");
 		adapter.setBeanFactory(mock(BeanFactory.class));
 		adapter.afterPropertiesSet();
@@ -106,7 +101,7 @@ public class DownstreamExceptionTests {
 	public void testWithErrorChannel() throws Exception {
 		assertThat(TestUtils.getPropertyValue(this.withErrorChannel, "errorChannel")).isSameAs(this.errors);
 		service.n = 0;
-		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
+		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler(MosquittoContainerTest.mqttUrl(), "si-test-out");
 		adapter.setDefaultTopic("mqtt-fooEx2");
 		adapter.setBeanFactory(mock(BeanFactory.class));
 		adapter.afterPropertiesSet();

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

@@ -36,4 +36,9 @@ public interface MosquittoContainerTest {
 					.withReuse(true)
 					.withExposedPorts(1883);
 
+
+	static String mqttUrl() {
+		return "tcp://localhost:" + MOSQUITTO_CONTAINER.getFirstMappedPort();
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -25,9 +25,7 @@ import javax.management.MalformedObjectNameException;
 import javax.management.ObjectName;
 
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -47,19 +45,16 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Artem Bilan
  *
  * @since 5.1.2
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
-public class MqttDslTests {
-
-	@ClassRule
-	public static final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883);
+public class MqttDslTests implements MosquittoContainerTest {
 
 	@Autowired
 	@Qualifier("mqttOutFlow.input")
@@ -107,7 +102,7 @@ public class MqttDslTests {
 		public DefaultMqttPahoClientFactory pahoClientFactory() {
 			DefaultMqttPahoClientFactory pahoClientFactory = new DefaultMqttPahoClientFactory();
 			MqttConnectOptions connectionOptions = new MqttConnectOptions();
-			connectionOptions.setServerURIs(new String[] { "tcp://localhost:1883" });
+			connectionOptions.setServerURIs(new String[]{ MosquittoContainerTest.mqttUrl() });
 			pahoClientFactory.setConnectionOptions(connectionOptions);
 			return pahoClientFactory;
 		}
@@ -120,8 +115,8 @@ public class MqttDslTests {
 		@Bean
 		public IntegrationFlow mqttInFlow() {
 			return IntegrationFlows.from(
-					new MqttPahoMessageDrivenChannelAdapter("jmxTestIn",
-							pahoClientFactory(), "jmxTests"))
+							new MqttPahoMessageDrivenChannelAdapter("jmxTestIn",
+									pahoClientFactory(), "jmxTests"))
 					.channel(c -> c.queue("fromMqttChannel"))
 					.get();
 		}

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

@@ -109,8 +109,6 @@ public class Mqttv5BackToBackTests implements MosquittoContainerTest {
 	@EnableIntegration
 	public static class Config {
 
-		private static final String MQTT_URL = "tcp://localhost:" + MOSQUITTO_CONTAINER.getFirstMappedPort();
-
 		List<MqttIntegrationEvent> events = new ArrayList<>();
 
 		@EventListener
@@ -147,7 +145,7 @@ public class Mqttv5BackToBackTests implements MosquittoContainerTest {
 		@Bean
 		public IntegrationFlow mqttOutFlow() {
 			Mqttv5PahoMessageHandler messageHandler =
-					new Mqttv5PahoMessageHandler(MQTT_URL, "mqttv5SIout");
+					new Mqttv5PahoMessageHandler(MosquittoContainerTest.mqttUrl(), "mqttv5SIout");
 			MqttHeaderMapper mqttHeaderMapper = new MqttHeaderMapper();
 			mqttHeaderMapper.setOutboundHeaderNames("foo", MessageHeaders.CONTENT_TYPE);
 			messageHandler.setHeaderMapper(mqttHeaderMapper);
@@ -161,7 +159,7 @@ public class Mqttv5BackToBackTests implements MosquittoContainerTest {
 		@Bean
 		public IntegrationFlow mqttInFlow() {
 			Mqttv5PahoMessageDrivenChannelAdapter messageProducer =
-					new Mqttv5PahoMessageDrivenChannelAdapter(MQTT_URL, "mqttv5SIin", "siTest");
+					new Mqttv5PahoMessageDrivenChannelAdapter(MosquittoContainerTest.mqttUrl(), "mqttv5SIin", "siTest");
 			messageProducer.setPayloadType(String.class);
 			messageProducer.setMessageConverter(mqttStringToBytesConverter());
 			messageProducer.setManualAcks(true);

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,8 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 import java.util.Collection;
 import java.util.Iterator;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
@@ -31,15 +30,16 @@ import org.springframework.integration.mqtt.support.MqttMessageConverter;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 4.0
  *
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class MqttMessageDrivenChannelAdapterParserTests {
 

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,8 +18,7 @@ package org.springframework.integration.mqtt.config.xml;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.aop.framework.Advised;
 import org.springframework.aop.support.AopUtils;
@@ -36,7 +35,7 @@ import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Gary Russell
@@ -44,17 +43,20 @@ import org.springframework.test.context.junit4.SpringRunner;
  *
  * @since 4.0
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
 public class MqttOutboundChannelAdapterParserTests {
 
-	@Autowired @Qualifier("withConverter")
+	@Autowired
+	@Qualifier("withConverter")
 	private EventDrivenConsumer withConverterEndpoint;
 
-	@Autowired @Qualifier("withConverter.handler")
+	@Autowired
+	@Qualifier("withConverter.handler")
 	private MessageHandler withConverterHandler;
 
-	@Autowired @Qualifier("withDefaultConverter.handler")
+	@Autowired
+	@Qualifier("withDefaultConverter.handler")
 	private MqttPahoMessageHandler withDefaultConverterHandler;
 
 	@Autowired