INT-4563: Add JMX test for MQTT

JIRA: https://jira.spring.io/browse/INT-4563

* Add a `MqttDslTests` with the JMX configured to be sure that SI
managed components are registered in JMX properly.
* Also this test covers a Java DSL configuration for MQTT channel adapters
* Some polishing for `AbstractMqttMessageHandler`
* Document Java DSL configuration for the MQTT channel adapters

build.gradle

@@ -540,6 +540,8 @@ project('spring-integration-mqtt') {
 	dependencies {
 		compile project(":spring-integration-core")
 		compile "org.eclipse.paho:org.eclipse.paho.client.mqttv3:$pahoMqttClientVersion"
+
+		testCompile project(":spring-integration-jmx")
 	}
 }
 

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java

@@ -44,7 +44,7 @@ import org.springframework.util.Assert;
 public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler implements Lifecycle {
 
 	private static final MessageProcessor<String> DEFAULT_TOPIC_PROCESSOR =
-			m -> (String) m.getHeaders().get(MqttHeaders.TOPIC);
+			m -> m.getHeaders().get(MqttHeaders.TOPIC, String.class);
 
 	private final AtomicBoolean running = new AtomicBoolean();
 
@@ -265,7 +265,7 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
 			throw new MessageHandlingException(message,
 					"No topic could be determined from the message and no default topic defined");
 		}
-		this.publish(topic == null ? this.defaultTopic : topic, mqttMessage, message);
+		publish(topic == null ? this.defaultTopic : topic, mqttMessage, message);
 	}
 
 	protected abstract void publish(String topic, Object mqttMessage, Message<?> message) throws Exception;

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

@@ -0,0 +1,131 @@
+/*
+ * Copyright 2018 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.mqtt;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Set;
+
+import javax.management.MBeanServer;
+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.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.jmx.config.EnableIntegrationMBeanExport;
+import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
+import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
+import org.springframework.integration.mqtt.support.MqttHeaders;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.jmx.support.MBeanServerFactoryBean;
+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;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 5.1.2
+ */
+@RunWith(SpringRunner.class)
+@DirtiesContext
+public class MqttDslTests {
+
+	@ClassRule
+	public static final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883);
+
+	@Autowired
+	@Qualifier("mqttOutFlow.input")
+	private MessageChannel mqttOutFlowInput;
+
+	@Autowired
+	private PollableChannel fromMqttChannel;
+
+	@Autowired
+	private MBeanServer server;
+
+	@Test
+	public void testMqttChannelAdaptersAndJmx() throws MalformedObjectNameException {
+		Set<ObjectName> mbeanNames = this.server.queryNames(
+				new ObjectName("org.springframework.integration:type=ManagedEndpoint,*"), null);
+
+		assertThat(mbeanNames.size()).isEqualTo(1);
+		ObjectName objectName = mbeanNames.iterator().next();
+		assertThat(objectName.toString()).contains("name=\"mqttInFlow.mqtt:inbound-channel-adapter#0\"");
+
+		String testPayload = "foo";
+
+		this.mqttOutFlowInput.send(
+				MessageBuilder.withPayload(testPayload)
+						.setHeader(MqttHeaders.TOPIC, "jmxTests")
+						.build());
+
+		Message<?> receive = this.fromMqttChannel.receive(10_000);
+
+		assertThat(receive).isNotNull();
+		assertThat(receive.getPayload()).isEqualTo(testPayload);
+	}
+
+	@Configuration
+	@EnableIntegration
+	@EnableIntegrationMBeanExport(server = "mbeanServer")
+	public static class Config {
+
+		@Bean
+		public static MBeanServerFactoryBean mbeanServer() {
+			return new MBeanServerFactoryBean();
+		}
+
+		@Bean
+		public DefaultMqttPahoClientFactory pahoClientFactory() {
+			DefaultMqttPahoClientFactory pahoClientFactory = new DefaultMqttPahoClientFactory();
+			MqttConnectOptions connectionOptions = new MqttConnectOptions();
+			connectionOptions.setServerURIs(new String[] { "tcp://localhost:1883" });
+			pahoClientFactory.setConnectionOptions(connectionOptions);
+			return pahoClientFactory;
+		}
+
+		@Bean
+		public IntegrationFlow mqttOutFlow() {
+			return f -> f.handle(new MqttPahoMessageHandler("jmxTestOut", pahoClientFactory()));
+		}
+
+		@Bean
+		public IntegrationFlow mqttInFlow() {
+			return IntegrationFlows.from(
+					new MqttPahoMessageDrivenChannelAdapter("jmxTestIn",
+							pahoClientFactory(), "jmxTests"))
+					.channel(c -> c.queue("fromMqttChannel"))
+					.get();
+		}
+
+	}
+
+}

src/reference/asciidoc/mqtt.adoc

@@ -80,7 +80,7 @@ The following listing shows the available attributes:
 <1> The client ID.
 <2> The broker URL.
 <3> A comma-separated list of topics from which this adapter receives messages.
-<4> A commap-separated list of QoS values.
+<4> A comma-separated list of QoS values.
 It can be a single value that is applied to all topics or a value for each topic (in which case, the lists must be the same length).
 <5> An `MqttMessageConverter` (optional).
 By default, the default `DefaultPahoMessageConverter` produces a message with a `String` payload with the following headers:
@@ -102,13 +102,11 @@ NOTE: Starting with version 4.1, you can omit the URL.
 Instead, you can provide the server URIs in the `serverURIs` property of the `DefaultMqttPahoClientFactory`.
 Doing so enables, for example, connection to a highly available (HA) cluster.
 
-Starting with version 4.2.2, an `MqttSubscribedEvent` is published when the adapter successfully subscribes to the
-topics.
+Starting with version 4.2.2, an `MqttSubscribedEvent` is published when the adapter successfully subscribes to the topics.
 `MqttConnectionFailedEvent` events are published when the connection or subscription fails.
 These events can be received by a bean that implements `ApplicationListener`.
 
-Also, a new property called `recoveryInterval` controls the interval at which the adapter attempts to reconnect after
-a failure.
+Also, a new property called `recoveryInterval` controls the interval at which the adapter attempts to reconnect after a failure.
 It defaults to `10000ms` (ten seconds).
 
 [NOTE]
@@ -196,6 +194,35 @@ public class MqttJavaApplication {
 ----
 ====
 
+==== Configuring with the Java DSL
+
+The following Spring Boot application provides an example of configuring the inbound adapter with the Java DSL:
+
+====
+[source, java]
+----
+@SpringBootApplication
+public class MqttJavaApplication {
+
+    public static void main(String[] args) {
+        new SpringApplicationBuilder(MqttJavaApplication.class)
+            .web(false)
+            .run(args);
+    }
+
+    @Bean
+    public IntegrationFlow mqttInbound() {
+        return IntegrationFlows.from(
+                         new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883",
+                                        "testClient", "topic1", "topic2");)
+                .handle(m -> System.out.println(m.getPayload()))
+                .get();
+    }
+
+}
+----
+====
+
 [[mqtt-outbound]]
 === Outbound Channel Adapter
 
@@ -315,3 +342,28 @@ public class MqttJavaApplication {
 }
 ----
 ====
+
+==== Configuring with the Java DSL
+
+The following Spring Boot application provides an example of configuring the outbound adapter with the Java DSL:
+
+====
+[source, java]
+----
+@SpringBootApplication
+public class MqttJavaApplication {
+
+    public static void main(String[] args) {
+        new SpringApplicationBuilder(MqttJavaApplication.class)
+            .web(false)
+            .run(args);
+    }
+
+   	@Bean
+   	public IntegrationFlow mqttOutboundFlow() {
+   	    return f -> f.handle(new MqttPahoMessageHandler("tcp://host1:1883", "someMqttClient"));
+    }
+
+}
+----
+====