mqtt.md

# MQTT Support 

## 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.5.9</version>
</dependency>
```

Gradle

```
compile "org.springframework.integration:spring-integration-mqtt:5.5.9"
```

The current implementation uses the [Eclipse Paho MQTT Client](https://www.eclipse.org/paho/) library.

|   |The XML configuration and most of this chapter are about MQTT v3.1 protocol support and respective Paho Client.<br/>See [MQTT v5 Support](#mqtt-v5) 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.|
|---|-------------------------------------------------------------------------------------------------------------------------------------------------------------|

###  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.<br/>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).<br/>By default, the default `DefaultPahoMessageConverter` produces a message with a `String` payload with the following headers:<br/><br/>* `mqtt_topic`: The topic from which the message was received<br/><br/>* `mqtt_duplicate`: `true` if the message is a duplicate<br/><br/>* `mqtt_qos`: The quality of service<br/>  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.<br/>It applies only if the channel might block (such as a bounded `QueueChannel` that is currently full).                                                                                                                                                                                                             |
|**8** |                                                                                                                                                                           The error channel.<br/>Downstream exceptions are sent to this channel, if supplied, in an `ErrorMessage`.<br/>The payload is a `MessagingException` that contains the failed message and cause.                                                                                                                                                                           |
|**9** |                                                                                                                                                                                             The recovery interval.<br/>It controls the interval at which the adapter attempts to reconnect after a failure.<br/>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.<br/>Instead, you can provide the server URIs in the `serverURIs` property of the `DefaultMqttPahoClientFactory`.<br/>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.<br/>This was incorrect because, if the client QOS is greater than 0, we need to keep the subscription active so that messages arriving<br/>while the adapter is stopped are delivered on the next start.<br/>This also requires setting the `cleanSession` property on the client factory to `false`.<br/>It defaults to `true`.<br/><br/>Starting with version 4.2.3, the adapter does not unsubscribe (by default) if the `cleanSession` property is `false`.<br/><br/>This behavior can be overridden by setting the `consumerCloseAction` property on the factory.<br/>It can have values: `UNSUBSCRIBE_ALWAYS`, `UNSUBSCRIBE_NEVER`, and `UNSUBSCRIBE_CLEAN`.<br/>The latter (the default) unsubscribes only if the `cleanSession` property is `true`.<br/><br/>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 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).<br/>The default `DefaultPahoMessageConverter` recognizes the following headers:<br/><br/>* `mqtt_topic`: The topic to which the message will be sent<br/><br/>* `mqtt_retained`: `true` if the message is to be retained<br/><br/>* `mqtt_qos`: The quality of service                                                                                                                                                                                                                                    |
|**4** |                                                                                                                                                                                                                                                                                                                                                                                The client factory.                                                                                                                                                                                                                                                                                                                                                                                 |
|**5** |                                                                                                                                                                                                                                                                                                   The default quality of service.<br/>It is used if no `mqtt_qos` header is found or the `qos-expression` returns `null`.<br/>It is not used if you supply a custom `converter`.                                                                                                                                                                                                                                                                                                   |
|**6** |                                                                                                                                                                                                                                                                                                                                              An expression to evaluate to determine the qos.<br/>The default is `headers[mqtt_qos]`.                                                                                                                                                                                                                                                                                                                                               |
|**7** |                                                                                                                                                                                                                                                                                                               The default value of the retained flag.<br/>It is used if no `mqtt_retained` header is found.<br/>It is not used if a custom `converter` is supplied.                                                                                                                                                                                                                                                                                                                |
|**8** |                                                                                                                                                                                                                                                                                                                                     An expression to evaluate to determine the retained boolean.<br/>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.<br/>The default is `headers['mqtt_topic']`.                                                                                                                                                                                                                                                                                                                                      |
|**11**|                                                                                                                                                                                                                                                                                              When `true`, the caller does not block.<br/>Rather, it waits for delivery confirmation when a message is sent.<br/>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](#mqtt-events)).<br/>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).<br/>When the delivery is confirmed by the client library, an `MqttMessageDeliveredEvent` is emitted.<br/>It contains the `messageId`, the `clientId`, and the `clientInstance`, enabling delivery to be correlated with the send.<br/>Any `ApplicationListener` or an event inbound channel adapter can received these events.<br/>Note that it is possible for the `MqttMessageDeliveredEvent` to be received before the `MqttMessageSentEvent`.<br/>The default is `false`.|

|   |Starting with version 4.1, the URL can be omitted.<br/>Instead, the server URIs can be provided in the `serverURIs` property of the `DefaultMqttPahoClientFactory`.<br/>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();
```

### 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.|
|---|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|

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 IntegrationFlows.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.<br/>Otherwise, only the manual restart of these channel adapters can handle reconnects, e.g. via `MqttConnectionFailedEvent` handling on disconnection.|
|---|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|