diff --git a/build.gradle b/build.gradle index ec670037af..4158c7dfb8 100644 --- a/build.gradle +++ b/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") } } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java index 33c5d9a233..22761c9f6d 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java +++ b/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 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; diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttDslTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttDslTests.java new file mode 100644 index 0000000000..91cbe74dc2 --- /dev/null +++ b/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 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(); + } + + } + +} diff --git a/src/reference/asciidoc/mqtt.adoc b/src/reference/asciidoc/mqtt.adoc index 8db2d2c1d9..9d1a817a9a 100644 --- a/src/reference/asciidoc/mqtt.adoc +++ b/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")); + } + +} +---- +====