Spring Integration provides inbound and outbound channel adapters supporting the MQ Telemetry Transport (MQTT) protocol. The current implementation uses the Eclipse Paho MQTT Client library.
Configuration of both adapters is achieved using the DefaultMqttPahoClientFactory
.
Refer to the Paho documentation for more information about configuration options.
The inbound channel adapter is implemented by the MqttPahoMessageDrivenChannelAdapter
.
For convenience, it can be configured using the namespace.
A minimal configuration might be:
<bean id="clientFactory" class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory"> <property name="userName" value="${mqtt.username}"/> <property name="password" value="${mqtt.password}"/> </bean> <int-mqtt:message-driven-channel-adapter id="mqttInbound" client-id="${mqtt.default.client.id}.src" url="${mqtt.url}" topics="sometopic" client-factory="clientFactory" channel="output"/>
Attributes:
<int-mqtt:message-driven-channel-adapter id="oneTopicAdapter" client-id="foo" url="tcp://localhost:1883" topics="bar,baz" qos="1,2" converter="myConverter" client-factory="clientFactory" send-timeout="123" error-channel="errors" recovery-interval="10000" channel="out" />
The client id. | |
The broker URL. | |
A comma delimited list of topics from which this adapter will receive messages. | |
A comma delimited list of QoS values. Can be a single value that is applied to all topics, or a value for each topic (in which case the lists must the same length). | |
An | |
The client factory. | |
The send timeout - only applies if the channel might block (such as a bounded | |
The error channel - downstream exceptions will be sent to this channel, if supplied, in an | |
The recovery interval - controls the interval at which the adapter will attempt to reconnect after
a failure; it defaults to |
Note | |
---|---|
Starting with version 4.1 the url can be omitted and, instead, the server URIs can be provided in the |
Starting with version 4.2.2, an MqttSubscribedEvent
is published when the adapter successfully subscribes to the
topic(s).
MqttConnectionFailedEvent
s are published when the connection/subscription fails.
These events can be received by a bean that implements ApplicationListener
.
Also, a new property recoveryInterval
controls the interval at which the adapter will attempt to reconnect after
a failure; it defaults to 10000ms
(ten seconds).
Note | |
---|---|
Prior to version 4.2.3, the client always unsubscribed when the adapter was stopped.
This was incorrect because if the client QOS is > 0, we need to keep the subscription active so that messages arriving
while the adapter is stopped will be delivered on the next start.
This also requires setting the Starting with version 4.2.3, the adapter will not unsubscribe (by default) if the This behavior can be overridden by setting the To revert to the pre-4.2.3 behavior, use |
Important | |
---|---|
Starting with version 5.0, the |
Starting with version 4.1, it is possible to programmatically change the topics to which the adapter is subscribed.
Methods addTopic()
and removeTopic()
are provided.
When adding topics, you can optionally specify the QoS
(default: 1).
You can also modify the topics by sending an appropriate message to a <control-bus/>
with an appropriate payload: "myMqttAdapter.addTopic('foo', 1)"
.
Stopping/starting the adapter has no effect on the topic list (it does not revert to the original settings in the configuration). The changes are not retained beyond the life cycle of the application context; a new application context will revert to the configured settings.
Changing the topics while the adapter is stopped (or disconnected from the broker) will take effect the next time a connection is established.
The following Spring Boot application provides an example of configuring the inbound adapter using Java configuration:
@SpringBootApplication public class MqttJavaApplication { public static void main(String[] args) { new SpringApplicationBuilder(MqttJavaApplication.class) .web(false) .run(args); } @Bean public MessageChannel mqttInputChannel() { return new DirectChannel(); } @Bean public MessageProducer inbound() { MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "testClient", "topic1", "topic2"); adapter.setCompletionTimeout(5000); adapter.setConverter(new DefaultPahoMessageConverter()); adapter.setQos(1); adapter.setOutputChannel(mqttInputChannel()); return adapter; } @Bean @ServiceActivator(inputChannel = "mqttInputChannel") public MessageHandler handler() { return new MessageHandler() { @Override public void handleMessage(Message<?> message) throws MessagingException { System.out.println(message.getPayload()); } }; } }
The outbound channel adapter is implemented by the MqttPahoMessageHandler
which is wrapped in a ConsumerEndpoint
.
For convenience, it can be configured using the namespace.
Starting with version 4.1, the adapter supports asynchronous sends, avoiding blocking until the delivery is confirmed; application events can be emitted to enable applications to confirm delivery if desired.
Attributes:
<int-mqtt:outbound-channel-adapter id="withConverter" client-id="foo" url="tcp://localhost:1883" converter="myConverter" client-factory="clientFactory" default-qos="1" qos-expression="" default-retained="true" retained-expression="" default-topic="bar" topic-expression="" async="false" async-events="false" channel="target" />
The client id. | |
The broker URL. | |
An | |
The client factory. | |
The default quality of service (used if no | |
An expression to evaluate to determine the qos; default | |
The default value of the retained flag (used if no | |
An expression to evaluate to determine the retained boolean; default | |
The default topic to which the message will be sent (used if no | |
An expression to evaluate to determine the destination topic; default | |
When | |
When |
Note | |
---|---|
Starting with version 4.1 the url can be omitted and, instead, the server URIs can be provided in the |
The following Spring Boot application provides an example of configuring the outbound adapter using Java configuration:
@SpringBootApplication @IntegrationComponentScan public class MqttJavaApplication { public static void main(String[] args) { ConfigurableApplicationContext context = new SpringApplicationBuilder(MqttJavaApplication.class) .web(false) .run(args); MyGateway gateway = context.getBean(MyGateway.class); gateway.sendToMqtt("foo"); } @Bean public MqttPahoClientFactory mqttClientFactory() { DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); factory.setServerURIs("tcp://host1:1883", "tcp://host2:1883"); factory.setUserName("username"); factory.setPassword("password"); return factory; } @Bean @ServiceActivator(inputChannel = "mqttOutboundChannel") public MessageHandler mqttOutbound() { MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("testClient", mqttClientFactory()); messageHandler.setAsync(true); messageHandler.setDefaultTopic("testTopic"); return messageHandler; } @Bean public MessageChannel mqttOutboundChannel() { return new DirectChannel(); } @MessagingGateway(defaultRequestChannel = "mqttOutboundChannel") public interface MyGateway { void sendToMqtt(String data); } }