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>6.1.0-M2</version>
</dependency>

Gradle

compile "org.springframework.integration:spring-integration-mqtt:6.1.0-M2"

The current implementation uses the Eclipse Paho MQTT Client library.

The XML configuration and most of this chapter are about MQTT v3.1 protocol support and respective Paho Client. See MQTT v5 Support paragraph for respective protocol support.

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: `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).
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 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.

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 IMqttClient setManualAcks() and messageArrivedComplete() for more information. For convenience a header accessor is provided:

StaticMessageHeaderAccessor.acknowledgment(someMessage).acknowledge();

Starting with version 5.2.11, when the message converter throws an exception or returns null from the MqttMessage conversion, the MqttPahoMessageDrivenChannelAdapter sends an ErrorMessage into the errorChannel, if provided. Re-throws this conversion error otherwise into an MQTT client callback.

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 IntegrationFlow.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 (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 `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. For the MQTT v5 Paho client, this event is also emitted when the server performs a normal disconnection, in which case the cause of the lost connection is null.
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();

MQTT v5 Support

Starting with version 5.5.5, the spring-integration-mqtt module provides channel adapter implementations for the MQTT v5 protocol. The org.eclipse.paho:org.eclipse.paho.mqttv5.client is an optional dependency, so has to be included explicitly in the target project.

Since the MQTT v5 protocol supports extra arbitrary properties in an MQTT message, the MqttHeaderMapper implementation has been introduced to map to/from headers on publish and receive operations. By default, (via the * pattern) it maps all the received PUBLISH frame properties (including user properties). On the outbound side it maps this subset of headers for PUBLISH frame: contentType, mqtt_messageExpiryInterval, mqtt_responseTopic, mqtt_correlationData.

The outbound channel adapter for the MQTT v5 protocol is present as an Mqttv5PahoMessageHandler. It requires a clientId and MQTT broker URL or MqttConnectionOptions reference. It supports a MqttClientPersistence option, can be async and can emit MqttIntegrationEvent objects in that case (see asyncEvents option). If a request message payload is an org.eclipse.paho.mqttv5.common.MqttMessage, it is published as is via the internal IMqttAsyncClient. If the payload is byte[] it is used as is for the target MqttMessage payload to publish. If the payload is a String it is converted to byte[] to publish. The remaining use-cases are delegated to the provided MessageConverter which is a IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME ConfigurableCompositeMessageConverter bean from the application context. Note: the provided HeaderMapper<MqttProperties> is not used when the requested message payload is already an MqttMessage. The following Java DSL configuration sample demonstrates how to use this channel adapter in the integration flow:

@Bean
public IntegrationFlow mqttOutFlow() {
    Mqttv5PahoMessageHandler messageHandler = new Mqttv5PahoMessageHandler(MQTT_URL, "mqttv5SIout");
    MqttHeaderMapper mqttHeaderMapper = new MqttHeaderMapper();
    mqttHeaderMapper.setOutboundHeaderNames("some_user_header", MessageHeaders.CONTENT_TYPE);
    messageHandler.setHeaderMapper(mqttHeaderMapper);
    messageHandler.setAsync(true);
    messageHandler.setAsyncEvents(true);
    messageHandler.setConverter(mqttStringToBytesConverter());

    return f -> f.handle(messageHandler);
}

The org.springframework.integration.mqtt.support.MqttMessageConverter cannot be used with the Mqttv5PahoMessageHandler since its contract is aimed only for the MQTT v3 protocol.

If connection fails on start up or at runtime, the Mqttv5PahoMessageHandler tries to reconnect on the next message produced to this handler. If this manual reconnection fails, the connection is exception is thrown back to the caller. In this case the standard Spring Integration error handling procedure is applied, including request handler advices, e.g. retry or circuit breaker.

See more information in the Mqttv5PahoMessageHandler javadocs and its superclass.

The inbound channel adapter for the MQTT v5 protocol is present as an Mqttv5PahoMessageDrivenChannelAdapter. It requires a clientId and MQTT broker URL or MqttConnectionOptions reference, plus topics to which to subscribe and consume from. It supports a MqttClientPersistence option, which is in-memory by default. The expected payloadType (byte[] by default) can be configured and it is propagated to the provided SmartMessageConverter for conversion from byte[] of the received MqttMessage. If the manualAck option is set, then an IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK header is added to the message to produce as an instance of SimpleAcknowledgment. The HeaderMapper<MqttProperties> is used to map PUBLISH frame properties (including user properties) into the target message headers. Standard MqttMessage properties, such as qos, id, dup, retained, plus received topic are always mapped to headers. See MqttHeaders for more information.

The following Java DSL configuration sample demonstrates how to use this channel adapter in the integration flow:

@Bean
public IntegrationFlow mqttInFlow() {
    Mqttv5PahoMessageDrivenChannelAdapter messageProducer =
        new Mqttv5PahoMessageDrivenChannelAdapter(MQTT_URL, "mqttv5SIin", "siTest");
    messageProducer.setPayloadType(String.class);
    messageProducer.setMessageConverter(mqttStringToBytesConverter());
    messageProducer.setManualAcks(true);

    return IntegrationFlow.from(messageProducer)
            .channel(c -> c.queue("fromMqttChannel"))
            .get();
}

The org.springframework.integration.mqtt.support.MqttMessageConverter cannot be used with the Mqttv5PahoMessageDrivenChannelAdapter since its contract is aimed only for the MQTT v3 protocol.

See more information in the Mqttv5PahoMessageDrivenChannelAdapter javadocs and its superclass.

It is recommended to have the MqttConnectionOptions#setAutomaticReconnect(boolean) set to true to let an internal IMqttAsyncClient instance to handle reconnects. Otherwise, only the manual restart of Mqttv5PahoMessageDrivenChannelAdapter can handle reconnects, e.g. via MqttConnectionFailedEvent handling on disconnection.

Shared MQTT Client Support

If a single MQTT ClientID is required for several integrations, multiple MQTT client instances cannot be used because MQTT brokers may have a limitation on a number of connections per ClientID (typically, a single connection is allowed). For having a single client reused for different channel adapters, a org.springframework.integration.mqtt.core.ClientManager component may be used and passed to any channel adapter needed. It will manage MQTT connection lifecycle and do automatic reconnects if needed. Also, a custom connection options and MqttClientPersistence may be provided to the client manager just as currently it can be done for channel adapter components.

Note that both MQTT v5 and v3 channel adapters are supported.

The following Java DSL configuration sample demonstrates how to use this client manager in the integration flow:

@Bean
public ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager() {
    MqttConnectionOptions connectionOptions = new MqttConnectionOptions();
    connectionOptions.setServerURIs(new String[]{ "tcp://localhost:1883" });
    connectionOptions.setConnectionTimeout(30000);
    connectionOptions.setMaxReconnectDelay(1000);
    connectionOptions.setAutomaticReconnect(true);
    Mqttv5ClientManager clientManager = new Mqttv5ClientManager(connectionOptions, "client-manager-client-id-v5");
    clientManager.setPersistence(new MqttDefaultFilePersistence());
    return clientManager;
}

@Bean
public IntegrationFlow mqttInFlowTopic1(
        ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager) {

    Mqttv5PahoMessageDrivenChannelAdapter messageProducer =
        new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, "topic1");
    return IntegrationFlow.from(messageProducer)
            .channel(c -> c.queue("fromMqttChannel"))
            .get();
}

@Bean
public IntegrationFlow mqttInFlowTopic2(
        ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager) {

    Mqttv5PahoMessageDrivenChannelAdapter messageProducer =
        new Mqttv5PahoMessageDrivenChannelAdapter(clientManager, "topic2");
    return IntegrationFlow.from(messageProducer)
            .channel(c -> c.queue("fromMqttChannel"))
            .get();
}

@Bean
public IntegrationFlow mqttOutFlow(
        ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager) {

    return f -> f.handle(new Mqttv5PahoMessageHandler(clientManager));
}