MQTT v5 Support

documentation/src/main/docs/mqtt/mqtt.md

@@ -38,3 +38,36 @@ mp.messaging.incoming.[channel-name].connector=smallrye-mqtt
 mp.messaging.outgoing.[channel-name].connector=smallrye-mqtt
 ```
 
+## Protocol version
+
+The connector supports both MQTT 3.1.1 and MQTT 5.0. The protocol
+version is selected per channel through the `version` attribute:
+
+```properties
+# MQTT 3.1.1 (default)
+mp.messaging.incoming.prices.version=4
+
+# MQTT 5.0
+mp.messaging.incoming.prices.version=5
+```
+
+When `version=5` is used, the connector exposes the additional MQTT 5.0
+features described in the rest of this section:
+
+- Connect-time properties: `session-expiry-interval`, `receive-maximum`,
+  `topic-alias-maximum`.
+- Last Will properties: `will-topic`, `will-payload`, `will-qos`,
+  `will-retain`, plus the MQTT 5.0 specific
+  `will-content-type`, `will-response-topic` and `will-delay-interval`.
+  Both `will-topic` and `will-payload` must be set together, otherwise
+  the connector fails to start.
+- Subscription options: `no-local`, `retain-as-published`,
+  `retain-handling`.
+- Per-message properties (User Properties, Content-Type, Response Topic,
+  Correlation Data, Message Expiry Interval, Payload Format Indicator)
+  available on inbound and outbound messages.
+
+!!!note
+    These options are accepted but ignored by the broker when
+    `version=4` is used.
+

documentation/src/main/docs/mqtt/receiving-mqtt-messages.md

@@ -53,7 +53,63 @@ The MQTT Connector does not handle the deserialization and creates a
 
 ## Inbound Metadata
 
-The MQTT connector does not provide inbound metadata.
+Each incoming `Message` carries a
+`io.smallrye.reactive.messaging.mqtt.ReceivingMqttMessageMetadata`
+instance that exposes the MQTT publish frame:
+
+```java
+ReceivingMqttMessageMetadata metadata = message
+        .getMetadata(ReceivingMqttMessageMetadata.class)
+        .orElseThrow();
+
+String topic       = metadata.getTopic();
+MqttQoS qos        = metadata.getQosLevel();
+boolean retain     = metadata.isRetain();
+boolean duplicate  = metadata.isDuplicate();
+int messageId      = metadata.getMessageId();
+```
+
+When connected to a broker in MQTT 5.0 mode (`version=5`), the metadata
+also exposes the MQTT 5.0 message properties:
+
+| Accessor                       | Returns                                                |
+|--------------------------------|--------------------------------------------------------|
+| `getProperties()`              | The full `io.netty.handler.codec.mqtt.MqttProperties`  |
+| `getUserProperties()`          | The User Properties (`List<StringPair>`)               |
+| `getContentType()`             | The `Content-Type` property, or `null`                 |
+| `getResponseTopic()`           | The `Response Topic` property, or `null`               |
+| `getCorrelationData()`         | The `Correlation Data` bytes, or `null`                |
+| `getMessageExpiryInterval()`   | The message expiry interval in seconds, or `null`      |
+| `getPayloadFormatIndicator()`  | The payload format indicator (0=binary, 1=UTF-8)       |
+
+For convenience the `MqttMessage` interface itself exposes
+`getResponseTopic()` and `getCorrelationData()` directly, so that the
+typical request/response handling code does not need to dig into the
+metadata. See
+[Sending messages to MQTT](sending-messages-to-mqtt.md#mqtt-5-requestresponse)
+for an example of how to use them when replying.
+
+## MQTT 5.0 subscription options
+
+In addition to `qos`, three MQTT 5.0 subscription options can be set on
+the inbound channel:
+
+```properties
+mp.messaging.incoming.prices.version=5
+mp.messaging.incoming.prices.no-local=true
+mp.messaging.incoming.prices.retain-as-published=true
+mp.messaging.incoming.prices.retain-handling=2
+```
+
+- `no-local` (default `false`): when `true` the broker does not forward
+  to this subscription the messages that were published by the same
+  connection.
+- `retain-as-published` (default `false`): preserves the `retain` flag
+  on messages delivered to the subscriber instead of clearing it.
+- `retain-handling` (default `0`): controls how retained messages are
+  sent on subscribe — `0` send them on subscribe, `1` send them only if
+  the subscription does not already exist, `2` never send retained
+  messages on subscribe.
 
 ## Failure Management
 

documentation/src/main/docs/mqtt/sending-messages-to-mqtt.md

@@ -60,7 +60,76 @@ The `Message` sent to MQTT can have various payload types:
 
 ## Outbound Metadata
 
-The MQTT connector does not provide outbound metadata.
+You can attach
+`io.smallrye.reactive.messaging.mqtt.SendingMqttMessageMetadata` to any
+outgoing `Message` to override the topic, QoS, retain flag and — for
+MQTT 5.0 — to attach arbitrary `MqttProperties` to the publish:
+
+```java
+MqttProperties properties = new MqttProperties();
+properties.add(new MqttProperties.UserProperty("source", "sensor-1"));
+
+return MqttMessage.of("prices", payload, MqttQoS.AT_LEAST_ONCE, false, properties);
+```
+
+The `MqttMessage` interface provides several factory methods that cover
+the common cases:
+
+| Factory                                                                  | Use case                                                    |
+|--------------------------------------------------------------------------|-------------------------------------------------------------|
+| `MqttMessage.of(payload)`                                                | Publish on the channel's default topic                      |
+| `MqttMessage.of(topic, payload)`                                         | Override the topic                                          |
+| `MqttMessage.of(topic, payload, qos)`                                    | Override topic and QoS                                      |
+| `MqttMessage.of(topic, payload, qos, retain)`                            | Override topic, QoS and retain flag                         |
+| `MqttMessage.of(topic, payload, qos, retain, MqttProperties properties)` | Add MQTT 5.0 properties                                     |
+| `MqttMessage.of(topic, payload, MqttProperties properties)`              | Add MQTT 5.0 properties keeping default QoS and retain flag |
+| `MqttMessage.ofResponse(request, payload, qos[, properties])`            | Build a reply to an MQTT 5.0 request (see below)            |
+
+## MQTT 5.0 request/response
+
+MQTT 5.0 supports request/response interactions through the
+`Response Topic` and `Correlation Data` message properties. When the
+inbound channel is configured with `version=5`, the connector exposes
+those values via the incoming `MqttMessage`, and the
+`MqttMessage.ofResponse(...)` helper builds a reply that targets the
+correct topic and propagates the correlation data automatically.
+
+The following example handles a request and replies on the topic
+indicated by the requester, attaching two MQTT 5.0 User Properties to
+the response:
+
+``` java
+{{ insert('mqtt/outbound/MqttRequestResponseHandler.java') }}
+```
+
+Configuration:
+
+```properties
+mp.messaging.incoming.requests.connector=smallrye-mqtt
+mp.messaging.incoming.requests.host=mqtt
+mp.messaging.incoming.requests.version=5
+mp.messaging.incoming.requests.topic=requests
+
+mp.messaging.outgoing.responses.connector=smallrye-mqtt
+mp.messaging.outgoing.responses.host=mqtt
+mp.messaging.outgoing.responses.version=5
+# The topic attribute is not used: ofResponse() overrides it with the
+# Response Topic carried by the incoming request.
+```
+
+`MqttMessage.ofResponse(request, payload, qos, properties)`:
+
+- Reads the `Response Topic` property from `request` and uses it as
+  destination topic of the reply. If the request has no Response Topic
+  (for example because it was sent in MQTT 3.1.1) an
+  `IllegalArgumentException` is thrown.
+- Copies the `Correlation Data` of the request, if any, into the reply
+  so that the original requester can match request and response.
+- Does not mutate the `MqttProperties` instance passed by the caller —
+  it builds an internal copy with `Correlation Data` added.
+
+The two-argument variant `MqttMessage.ofResponse(request, payload, qos)`
+behaves the same way but uses an empty set of additional properties.
 
 ## Acknowledgement
 

documentation/src/main/java/mqtt/outbound/MqttRequestResponseHandler.java

@@ -0,0 +1,28 @@
+package mqtt.outbound;
+
+import jakarta.enterprise.context.ApplicationScoped;
+
+import org.eclipse.microprofile.reactive.messaging.Incoming;
+import org.eclipse.microprofile.reactive.messaging.Outgoing;
+
+import io.netty.handler.codec.mqtt.MqttProperties;
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.smallrye.reactive.messaging.mqtt.MqttMessage;
+
+@ApplicationScoped
+public class MqttRequestResponseHandler {
+
+    @Incoming("requests")
+    @Outgoing("responses")
+    public MqttMessage<byte[]> handle(MqttMessage<byte[]> request) {
+        // Process the request payload (request.getPayload()) here...
+
+        MqttProperties properties = new MqttProperties();
+        properties.add(new MqttProperties.UserProperty("version", "2"));
+        properties.add(new MqttProperties.UserProperty("result", "0"));
+
+        // ofResponse() targets the Response Topic carried by the request and
+        // copies the Correlation Data, so the requester can match the reply.
+        return MqttMessage.ofResponse(request, null, MqttQoS.AT_MOST_ONCE, properties);
+    }
+}

smallrye-reactive-messaging-mqtt/pom.xml

@@ -12,6 +12,16 @@
 
   <name>SmallRye Reactive Messaging : Connector :: MQTT</name>
 
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>io.vertx</groupId>
+        <artifactId>vertx-mqtt</artifactId>
+        <version>4.5.29-SNAPSHOT</version>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
+
   <dependencies>
     <dependency>
       <groupId>${project.groupId}</groupId>

smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/Clients.java

@@ -29,12 +29,20 @@ static ClientHolder getHolder(Vertx vertx, MqttClientSessionOptions options) {
         String clientId = Optional.ofNullable(options.getClientId()).orElse("");
         String server = options.getServerName().orElse("");
         String username = options.getUsername();
+        boolean ssl = options.isSsl();
+        boolean trustAll = options.isTrustAll();
+        String willTopic = Optional.ofNullable(options.getWillTopic()).orElse("");
+        int willQoS = options.getWillQoS();
+        boolean willRetain = options.isWillRetain();
+        int version = options.getVersion();
+
+        String id = String.format("%s@%s:%s<%s>-[%s]-ssl:%s-trustAll:%s-will:%s:%d:%s-v:%d",
+                username, host, port, server, clientId, ssl, trustAll, willTopic, willQoS, willRetain, version);
 
-        String id = String.format("%s@%s:%s<%s>-[%s]", username, host, port, server, clientId);
         return clients.computeIfAbsent(id, key -> {
             log.infof("Create MQTT Client for %s", id);
             MqttClientSession client = MqttClientSession.create(vertx.getDelegate(), options);
-            return new ClientHolder(client);
+            return new ClientHolder(client, options);
         });
     }
 
@@ -49,10 +57,12 @@ public static void clear() {
     public static class ClientHolder {
 
         private final MqttClientSession client;
+        private final MqttClientSessionOptions options;
         private final BroadcastProcessor<MqttPublishMessage> messages;
 
-        public ClientHolder(MqttClientSession client) {
+        public ClientHolder(MqttClientSession client, MqttClientSessionOptions options) {
             this.client = client;
+            this.options = options;
             messages = BroadcastProcessor.create();
             client.messageHandler(m -> messages.onNext(MqttPublishMessage.newInstance(m)));
         }

smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java

@@ -45,9 +45,10 @@
 @ConnectorAttribute(name = "keep-alive-seconds", type = "int", description = "Set the keep alive timeout in seconds", defaultValue = "30", direction = INCOMING_AND_OUTGOING)
 @ConnectorAttribute(name = "max-inflight-queue", type = "int", direction = INCOMING_AND_OUTGOING, description = "Set max count of unacknowledged messages", defaultValue = "10")
 @ConnectorAttribute(name = "auto-clean-session", type = "boolean", direction = INCOMING_AND_OUTGOING, description = "Set to start with a clean session (`true` by default)", defaultValue = "true")
-@ConnectorAttribute(name = "will-flag", type = "boolean", direction = INCOMING_AND_OUTGOING, description = "Set if will information are provided on connection", defaultValue = "false")
 @ConnectorAttribute(name = "will-retain", type = "boolean", direction = INCOMING_AND_OUTGOING, description = "Set if the will message must be retained", defaultValue = "false")
 @ConnectorAttribute(name = "will-qos", type = "int", direction = INCOMING_AND_OUTGOING, description = "Set the QoS level for the will message", defaultValue = "0")
+@ConnectorAttribute(name = "will-topic", type = "string", direction = INCOMING_AND_OUTGOING, description = "Set the will message's topic", defaultValue = "")
+@ConnectorAttribute(name = "will-payload", type = "string", direction = INCOMING_AND_OUTGOING, description = "Set the will message's payload", defaultValue = "")
 @ConnectorAttribute(name = "max-message-size", type = "int", direction = INCOMING_AND_OUTGOING, description = "Set max MQTT message size in bytes", defaultValue = "8092")
 @ConnectorAttribute(name = "reconnect-interval-seconds", type = "int", direction = INCOMING_AND_OUTGOING, description = "Set the reconnect interval in seconds", defaultValue = "1")
 @ConnectorAttribute(name = "username", type = "string", direction = INCOMING_AND_OUTGOING, description = "Set the username to connect to the server")
@@ -66,6 +67,16 @@
 @ConnectorAttribute(name = "buffer-size", direction = INCOMING, description = "The size buffer of incoming messages waiting to be processed", type = "int", defaultValue = "128")
 @ConnectorAttribute(name = "unsubscribe-on-disconnection", direction = INCOMING_AND_OUTGOING, description = "This flag restore the old behavior to unsubscribe from the broken on disconnection", type = "boolean", defaultValue = "false")
 @ConnectorAttribute(name = "retain", direction = OUTGOING, description = "Whether the published message should be retained", type = "boolean", defaultValue = "false")
+@ConnectorAttribute(name = "version", type = "int", direction = INCOMING_AND_OUTGOING, description = "MQTT protocol version: 4 for MQTT 3.1.1 (default), 5 for MQTT 5.0", defaultValue = "4")
+@ConnectorAttribute(name = "session-expiry-interval", type = "long", direction = INCOMING_AND_OUTGOING, description = "MQTT 5.0: Session Expiry Interval in seconds")
+@ConnectorAttribute(name = "receive-maximum", type = "int", direction = INCOMING_AND_OUTGOING, description = "MQTT 5.0: Receive Maximum - flow control window")
+@ConnectorAttribute(name = "topic-alias-maximum", type = "int", direction = INCOMING_AND_OUTGOING, description = "MQTT 5.0: Topic Alias Maximum advertised to broker", defaultValue = "255")
+@ConnectorAttribute(name = "will-content-type", type = "string", direction = INCOMING_AND_OUTGOING, description = "MQTT 5.0: Content-Type of the will payload")
+@ConnectorAttribute(name = "will-response-topic", type = "string", direction = INCOMING_AND_OUTGOING, description = "MQTT 5.0: Response Topic for the will message")
+@ConnectorAttribute(name = "will-delay-interval", type = "long", direction = INCOMING_AND_OUTGOING, description = "MQTT 5.0: Will Delay Interval in seconds")
+@ConnectorAttribute(name = "no-local", type = "boolean", direction = INCOMING, description = "MQTT 5.0: No-Local subscription option", defaultValue = "false")
+@ConnectorAttribute(name = "retain-as-published", type = "boolean", direction = INCOMING, description = "MQTT 5.0: Retain-As-Published subscription option", defaultValue = "false")
+@ConnectorAttribute(name = "retain-handling", type = "int", direction = INCOMING, description = "MQTT 5.0: Retain Handling (0=send on subscribe, 1=send if not yet, 2=never send)", defaultValue = "0")
 public class MqttConnector implements InboundConnector, OutboundConnector, HealthReporter {
 
     static final String CONNECTOR_NAME = "smallrye-mqtt";

smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttMessage.java

@@ -1,8 +1,11 @@
 package io.smallrye.reactive.messaging.mqtt;
 
+import static io.smallrye.reactive.messaging.mqtt.i18n.MqttExceptions.ex;
+
 import java.util.concurrent.CompletionStage;
 import java.util.function.Supplier;
 
+import io.netty.handler.codec.mqtt.MqttProperties;
 import io.netty.handler.codec.mqtt.MqttQoS;
 import io.smallrye.reactive.messaging.providers.locals.ContextAwareMessage;
 
@@ -36,9 +39,40 @@ static <T> MqttMessage<T> of(String topic, T payload, MqttQoS qos, boolean retai
         return new SendingMqttMessage<>(payload, new SendingMqttMessageMetadata(topic, qos, retain));
     }
 
+    static <T> MqttMessage<T> of(String topic, T payload, MqttQoS qos, boolean retain, MqttProperties properties) {
+        return new SendingMqttMessage<>(payload, new SendingMqttMessageMetadata(topic, qos, retain, properties));
+    }
+
+    static <T> MqttMessage<T> of(String topic, T payload, MqttProperties properties) {
+        return new SendingMqttMessage<>(payload, new SendingMqttMessageMetadata(topic, null, false, properties));
+    }
+
+    static <T> MqttMessage<T> ofResponse(MqttMessage<?> inputMessage, T payload, MqttQoS qos) {
+        return ofResponse(inputMessage, payload, qos, new MqttProperties());
+    }
+
+    static <T> MqttMessage<T> ofResponse(MqttMessage<?> inputMessage, T payload, MqttQoS qos, MqttProperties properties) {
+        String responseTopic = inputMessage.getResponseTopic();
+        if (responseTopic == null) {
+            throw ex.illegalArgumentMissingResponseTopic();
+        }
+        MqttProperties props = new MqttProperties();
+        properties.listAll().forEach(props::add);
+        byte[] correlationData = inputMessage.getCorrelationData();
+        if (correlationData != null) {
+            props.add(new MqttProperties.BinaryProperty(
+                    MqttProperties.MqttPropertyType.CORRELATION_DATA.value(), correlationData));
+        }
+        return new SendingMqttMessage<>(payload, new SendingMqttMessageMetadata(responseTopic, qos, false, props));
+    }
+
+    // TODO Should be removed?
     default MqttMessage<T> withAck(Supplier<CompletionStage<Void>> ack) {
-        return new SendingMqttMessage<>(getPayload(), new SendingMqttMessageMetadata(getTopic(), getQosLevel(), isRetain()),
-                ack);
+        MqttProperties props = getMetadata(SendingMqttMessageMetadata.class)
+                .map(SendingMqttMessageMetadata::getProperties)
+                .orElse(MqttProperties.NO_PROPERTIES);
+        return new SendingMqttMessage<>(getPayload(),
+                new SendingMqttMessageMetadata(getTopic(), getQosLevel(), isRetain(), props), ack);
     }
 
     int getMessageId();
@@ -50,4 +84,16 @@ default MqttMessage<T> withAck(Supplier<CompletionStage<Void>> ack) {
     boolean isRetain();
 
     String getTopic();
+
+    default String getResponseTopic() {
+        return getMetadata(ReceivingMqttMessageMetadata.class)
+                .map(ReceivingMqttMessageMetadata::getResponseTopic)
+                .orElse(null);
+    }
+
+    default byte[] getCorrelationData() {
+        return getMetadata(ReceivingMqttMessageMetadata.class)
+                .map(ReceivingMqttMessageMetadata::getCorrelationData)
+                .orElse(null);
+    }
 }