MQTT Support
Spring Integration provides inbound and outbound channel adapters to support the Message Queueing Telemetry Transport (MQTT) protocol.
You need to include this dependency into your project:
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-mqtt</artifactId>
<version>5.4.0-M3</version>
</dependency>compile "org.springframework.integration:spring-integration-mqtt:5.4.0-M3"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.
We recommend configuring an MqttConnectOptions object and injecting it into the factory, instead of setting the (deprecated) options on the factory itself.
|
Inbound (Message-driven) Channel Adapter
The inbound channel adapter is implemented by the MqttPahoMessageDrivenChannelAdapter.
For convenience, you can configure it by using the namespace.
A minimal configuration might be as follows:
<bean id="clientFactory"
class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
<property name="connectionOptions">
<bean class="org.eclipse.paho.client.mqttv3.MqttConnectOptions">
<property name="userName" value="${mqtt.username}"/>
<property name="password" value="${mqtt.password}"/>
</bean>
</property>
</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"/>The following listing shows the available attributes:
<int-mqtt:message-driven-channel-adapter id="oneTopicAdapter"
client-id="foo" (1)
url="tcp://localhost:1883" (2)
topics="bar,baz" (3)
qos="1,2" (4)
converter="myConverter" (5)
client-factory="clientFactory" (6)
send-timeout="123" (7)
error-channel="errors" (8)
recovery-interval="10000" (9)
manual-acks="false" (10)
channel="out" />| 1 | The client ID. |
| 2 | The broker URL. |
| 3 | A comma-separated list of topics from which this adapter receives messages. |
| 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:
|
| 6 | The client factory. |
| 7 | The send timeout.
It applies only if the channel might block (such as a bounded QueueChannel that is currently full). |
| 8 | The error channel.
Downstream exceptions are sent to this channel, if supplied, in an ErrorMessage.
The payload is a MessagingException that contains the failed message and cause. |
| 9 | The recovery interval.
It controls the interval at which the adapter attempts to reconnect after a failure.
It defaults to 10000ms (ten seconds). |
| 10 | The acknowledgment mode; set to true for manual acknowledgment. |
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.
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.
It defaults to 10000ms (ten seconds).
|
Prior to version 4.2.3, the client always unsubscribed when the adapter was stopped.
This was incorrect because, if the client QOS is greater than 0, we need to keep the subscription active so that messages arriving
while the adapter is stopped are delivered on the next start.
This also requires setting the Starting with version 4.2.3, the adapter does not unsubscribe (by default) if the This behavior can be overridden by setting the To revert to the pre-4.2.3 behavior, use |
|
Starting with version 5.0, the |
Adding and Removing Topics at Runtime
Starting with version 4.1, you can programmatically change the topics to which the adapter is subscribed.
Spring Integration provides the addTopic() and removeTopic() methods.
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 — for example: "myMqttAdapter.addTopic('foo', 1)".
Stopping and 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 reverts to the configured settings.
Changing the topics while the adapter is stopped (or disconnected from the broker) takes effect the next time a connection is established.
Manual Acks
Starting with version 5.3, you can set the manualAcks property to true.
Often used to asynchronously acknowledge delivery.
When set to true, header (IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK) is added to the message with the value being a SimpleAcknowledgment.
You must invoke the acknowledge() method to complete the delivery.
See the Javadocs for IMqppClient setManualAcks() and messageArrivedComplete() for more information.
For convenience a header accessor is provided:
StaticMessageHeaderAccessor.acknowledgment(someMessage).acknowledge();Configuring with Java Configuration
The following Spring Boot application shows an example of how to configure the inbound adapter with 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());
}
};
}
}Configuring with the Java DSL
The following Spring Boot application provides an example of configuring the inbound adapter with the Java DSL:
@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();
}
}Outbound Channel Adapter
The outbound channel adapter is implemented by the MqttPahoMessageHandler, which is wrapped in a ConsumerEndpoint.
For convenience, you can configure it by using the namespace.
Starting with version 4.1, the adapter supports asynchronous send operations, avoiding blocking until the delivery is confirmed. You can emit application events to enable applications to confirm delivery if desired.
The following listing shows the attributes available for an outbound channel adapter:
<int-mqtt:outbound-channel-adapter id="withConverter"
client-id="foo" (1)
url="tcp://localhost:1883" (2)
converter="myConverter" (3)
client-factory="clientFactory" (4)
default-qos="1" (5)
qos-expression="" (6)
default-retained="true" (7)
retained-expression="" (8)
default-topic="bar" (9)
topic-expression="" (10)
async="false" (11)
async-events="false" (12)
channel="target" />| 1 | The client ID. |
| 2 | The broker URL. |
| 3 | An MqttMessageConverter (optional).
The default DefaultPahoMessageConverter recognizes the following headers:
|
| 4 | The client factory. |
| 5 | The default quality of service.
It is used if no mqtt_qos header is found or the qos-expression returns null.
It is not used if you supply a custom converter. |
| 6 | An expression to evaluate to determine the qos.
The default is headers[mqtt_qos]. |
| 7 | The default value of the retained flag.
It is used if no mqtt_retained header is found.
It is not used if a custom converter is supplied. |
| 8 | An expression to evaluate to determine the retained boolean.
The default is headers[mqtt_retained]. |
| 9 | The default topic to which the message is sent (used if no mqtt_topic header is found). |
| 10 | An expression to evaluate to determine the destination topic.
The default is headers['mqtt_topic']. |
| 11 | When true, the caller does not block.
Rather, it waits for delivery confirmation when a message is sent.
The default is false (the send blocks until delivery is confirmed). |
| 12 | When async and async-events are both true, an MqttMessageSentEvent is emitted (See Events).
It contains the message, the topic, the messageId generated by the client library, the clientId, and the clientInstance (incremented each time the client is connected).
When the delivery is confirmed by the client library, an MqttMessageDeliveredEvent is emitted.
It contains the the messageId, the clientId, and the clientInstance, enabling delivery to be correlated with the send.
Any ApplicationListener or an event inbound channel adapter can received these events.
Note that it is possible for the MqttMessageDeliveredEvent to be received before the MqttMessageSentEvent.
The default is false. |
Starting with version 4.1, the URL can be omitted.
Instead, the server URIs can be provided in the serverURIs property of the DefaultMqttPahoClientFactory.
This enables, for example, connection to a highly available (HA) cluster.
|
Configuring with Java Configuration
The following Spring Boot application show an example of how to configure the outbound adapter with 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();
MqttConnectOptions options = new MqttConnectOptions();
options.setServerURIs(new String[] { "tcp://host1:1883", "tcp://host2:1883" });
options.setUserName("username");
options.setPassword("password".toCharArray());
factory.setConnectionOptions(options);
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);
}
}Configuring with the Java DSL
The following Spring Boot application provides an example of configuring the outbound adapter with the Java DSL:
@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"));
}
}Events
Certain application events are published by the adapters.
-
MqttConnectionFailedEvent- published by both adapters if we fail to connect or a connection is subsequently lost. -
MqttMessageSentEvent- published by the outbound adapter when a message has been sent, if running in asynchronous mode. -
MqttMessageDeliveredEvent- published by the outbound adapter when the client indicates that a message has been delivered, if running in asynchronous mode. -
MqttSubscribedEvent- published by the inbound adapter after subscribing to the topics.
These events can be received by an ApplicationListener<MqttIntegrationEvent> or with an @EventListener method.
To determine the source of an event, use the following; you can check the bean name and/or the connect options (to access the server URIs etc).
MqttPahoComponent source = event.getSourceAsType();
String beanName = source.getBeanName();
MqttConnectOptions options = source.getConnectionInfo();