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:

Maven

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
    <version>5.3.0.M1</version>
</dependency>

Gradle

compile "org.springframework.integration:spring-integration-mqtt:5.3.0.M1"

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)
    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: `mqtt_topic`: The topic from which the message was received `mqtt_duplicate`: `true` if the message is a duplicate `mqtt_qos`: The quality of service You can configure the `DefaultPahoMessageConverter` to return the raw `byte[]` in the payload by declaring it as a `<bean/>` and setting the `payloadAsBytes` property to `true`.
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).

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 cleanSession property on the client factory to false. It defaults to true.

Starting with version 4.2.3, the adapter does not unsubscribe (by default) if the cleanSession property is false.

This behavior can be overridden by setting the consumerCloseAction property on the factory. It can have values: UNSUBSCRIBE_ALWAYS, UNSUBSCRIBE_NEVER, and UNSUBSCRIBE_CLEAN. The latter (the default) unsubscribes only if the cleanSession property is true.

To revert to the pre-4.2.3 behavior, use UNSUBSCRIBE_ALWAYS.

Starting with version 5.0, the topic, qos, and retained properties are mapped to .RECEIVED_… headers (MqttHeaders.RECEIVED_TOPIC, MqttHeaders.RECEIVED_QOS, and MqttHeaders.RECEIVED_RETAINED), to avoid inadvertent propagation to an outbound message that (by default) uses the MqttHeaders.TOPIC, MqttHeaders.QOS, and MqttHeaders.RETAINED headers.

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.

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: `mqtt_topic`: The topic to which the message will be sent `mqtt_retained`: `true` if the message is to be retained `mqtt_qos`: The quality of service
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. 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"));
    }

}