MQTT: Fix deprecations; add test

Polishing - PR Comments

Polishing - PR Comments.

* Fix Copyright and `@since` in the affected classes

basic/mqtt/pom.xml

@@ -145,6 +145,28 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <dependency>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-starter-test</artifactId>
+      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <artifactId>jackson-module-kotlin</artifactId>
+          <groupId>com.fasterxml.jackson.module</groupId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.integration</groupId>
+      <artifactId>spring-integration-test</artifactId>
+      <scope>test</scope>
+      <exclusions>
+        <exclusion>
+          <artifactId>jackson-module-kotlin</artifactId>
+          <groupId>com.fasterxml.jackson.module</groupId>
+        </exclusion>
+      </exclusions>
+    </dependency>
   </dependencies>
   <repositories>
     <repository>

basic/mqtt/src/test/java/org/springframework/integration/samples/mqtt/ApplicationTest.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 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
+ *
+ *      http://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.samples.mqtt;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.springframework.integration.test.mock.MockIntegration.messageArgumentCaptor;
+import static org.springframework.integration.test.mock.MockIntegration.mockMessageHandler;
+
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.test.context.MockIntegrationContext;
+import org.springframework.integration.test.context.SpringIntegrationTest;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author Gary Russell
+ *
+ * @since 5.2
+ *
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest
+@SpringIntegrationTest
+public class ApplicationTest {
+
+	@ClassRule
+	public static final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883);
+
+	@Autowired
+	private MockIntegrationContext mockIntegrationContext;
+
+	@Autowired
+	private IntegrationFlow mqttOutFlow;
+
+	@Test
+	public void test() {
+		ArgumentCaptor<Message<?>> captor = messageArgumentCaptor();
+		MessageHandler mockMessageHandler = mockMessageHandler(captor).handleNext(m -> { });
+		this.mockIntegrationContext
+				.substituteMessageHandlerFor(
+						"mqttInFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#1",
+						mockMessageHandler);
+		this.mqttOutFlow.getInputChannel().send(new GenericMessage<>("foo"));
+		verify(mockMessageHandler).handleMessage(any());
+		assertThat(captor.getValue().getPayload())
+				.isEqualTo("foo sent to MQTT, received from MQTT");
+	}
+
+}

basic/mqtt/src/test/java/org/springframework/integration/samples/mqtt/BrokerRunning.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 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
+ *
+ *      http://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.samples.mqtt;
+
+import static org.junit.Assume.assumeNoException;
+import static org.junit.Assume.assumeTrue;
+
+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
+ *
+ * @since 5.2
+ *
+ */
+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) {
+		assumeTrue(brokerOnline.get(port));
+		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);
+			assumeNoException(e);
+		}
+		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);
+	}
+
+}

build.gradle

@@ -662,6 +662,8 @@ project('mqtt') {
 		compile 'org.springframework.boot:spring-boot-starter-integration'
 		compile "org.springframework.integration:spring-integration-stream"
 		compile "org.springframework.integration:spring-integration-mqtt"
+		testCompile 'org.springframework.boot:spring-boot-starter-test'
+		testCompile "org.springframework.integration:spring-integration-test"
 	}
 
 	bootRun {