From a1f438d28b6462200841427cc515938c71fc8b3d Mon Sep 17 00:00:00 2001 From: Domenico Briganti Date: Sun, 3 May 2026 10:46:09 +0200 Subject: [PATCH 01/11] Fix typo in MQTT channel will-qos config key --- .../smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java index 2b9b01d617..b93d03ffc3 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java @@ -203,7 +203,7 @@ static void merge(MqttClientSessionOptions custom, MqttConnectorCommonConfigurat custom.setTrustAll(config.getTrustAll()); } - if (isSetInChannelConfiguration("will-qus", config)) { + if (isSetInChannelConfiguration("will-qos", config)) { custom.setWillQoS(config.getWillQos()); } From 682886a6bb448e2a6829dc8bd470af8af40410e2 Mon Sep 17 00:00:00 2001 From: Domenico Briganti Date: Sun, 3 May 2026 10:47:32 +0200 Subject: [PATCH 02/11] Expose MQTT will-topic and will-payload connector attributes --- .../smallrye/reactive/messaging/mqtt/MqttConnector.java | 2 ++ .../reactive/messaging/mqtt/internal/MqttHelpers.java | 9 +++++++++ 2 files changed, 11 insertions(+) diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java index b1f8e32993..f466d23efe 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java @@ -48,6 +48,8 @@ @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 topic of the will message", defaultValue = "") +@ConnectorAttribute(name = "will-payload", type = "string", direction = INCOMING_AND_OUTGOING, description = "Set the payload of the will message", 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") diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java index b93d03ffc3..c984ff8421 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java @@ -15,6 +15,7 @@ import io.smallrye.reactive.messaging.mqtt.session.ConstantReconnectDelayOptions; import io.smallrye.reactive.messaging.mqtt.session.MqttClientSessionOptions; import io.smallrye.reactive.messaging.mqtt.session.ReconnectDelayOptions; +import io.vertx.core.buffer.Buffer; import io.vertx.core.net.JksOptions; import io.vertx.core.net.KeyCertOptions; import io.vertx.core.net.PemKeyCertOptions; @@ -59,6 +60,8 @@ private static MqttClientSessionOptions createMqttClientOptions(MqttConnectorCom options.setWillQoS(config.getWillQos()); options.setWillFlag(config.getWillFlag()); options.setWillRetain(config.getWillRetain()); + options.setWillTopic(config.getWillTopic()); + options.setWillMessageBytes(Buffer.buffer(config.getWillPayload())); options.setUnsubscribeOnDisconnect(config.getUnsubscribeOnDisconnection()); options.setMetricsName("mqtt|" + config.getChannel()); @@ -214,6 +217,12 @@ static void merge(MqttClientSessionOptions custom, MqttConnectorCommonConfigurat if (isSetInChannelConfiguration("will-retain", config)) { custom.setWillRetain(config.getWillRetain()); } + if (isSetInChannelConfiguration("will-topic", config)) { + custom.setWillTopic(config.getWillTopic()); + } + if (isSetInChannelConfiguration("will-payload", config)) { + custom.setWillMessageBytes(Buffer.buffer(config.getWillPayload())); + } if (isSetInChannelConfiguration("unsubscribe-on-disconnection", config)) { custom.setUnsubscribeOnDisconnect(config.getUnsubscribeOnDisconnection()); } From c764eee5f2f898480a4690f7c79467ddf6f45712 Mon Sep 17 00:00:00 2001 From: Domenico Briganti Date: Fri, 24 Apr 2026 11:50:05 +0200 Subject: [PATCH 03/11] Start working on MQTTv5 --- smallrye-reactive-messaging-mqtt/pom.xml | 10 ++++++++++ .../reactive/messaging/mqtt/internal/MqttHelpers.java | 2 ++ 2 files changed, 12 insertions(+) diff --git a/smallrye-reactive-messaging-mqtt/pom.xml b/smallrye-reactive-messaging-mqtt/pom.xml index 3bd408dfb7..e8816b228a 100644 --- a/smallrye-reactive-messaging-mqtt/pom.xml +++ b/smallrye-reactive-messaging-mqtt/pom.xml @@ -12,6 +12,16 @@ SmallRye Reactive Messaging : Connector :: MQTT + + + + io.vertx + vertx-mqtt + 4.5.27-SNAPSHOT + + + + ${project.groupId} diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java index c984ff8421..c55b4c1717 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java @@ -65,6 +65,8 @@ private static MqttClientSessionOptions createMqttClientOptions(MqttConnectorCom options.setUnsubscribeOnDisconnect(config.getUnsubscribeOnDisconnection()); options.setMetricsName("mqtt|" + config.getChannel()); + config.config().getOptionalValue("version", Integer.class).ifPresent(v -> options.setVersion(v)); + return options; } From 03b90985bea5d780cd70d23da408ea8514cfa9e8 Mon Sep 17 00:00:00 2001 From: Domenico Briganti Date: Sun, 3 May 2026 09:55:20 +0200 Subject: [PATCH 04/11] Implement MQTT 5.0 features including subscription options and message properties --- .../reactive/messaging/mqtt/Clients.java | 15 +++- .../messaging/mqtt/MqttConnector.java | 14 ++- .../reactive/messaging/mqtt/MqttMessage.java | 17 +++- .../reactive/messaging/mqtt/MqttSink.java | 7 +- .../reactive/messaging/mqtt/MqttSource.java | 22 ++++- .../mqtt/ReceivingMqttMessageMetadata.java | 68 +++++++++++++++ .../mqtt/SendingMqttMessageMetadata.java | 11 +++ .../messaging/mqtt/i18n/MqttExceptions.java | 3 + .../messaging/mqtt/internal/MqttHelpers.java | 86 +++++++++++++++---- .../mqtt/session/MqttClientSession.java | 25 ++++++ .../session/MqttClientSessionOptions.java | 27 ++++++ .../session/impl/MqttClientSessionImpl.java | 79 +++++++++++------ .../mqtt/session/impl/SubscriptionEntry.java | 18 ++++ 13 files changed, 335 insertions(+), 57 deletions(-) create mode 100644 smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/impl/SubscriptionEntry.java diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/Clients.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/Clients.java index faba140570..c990060048 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/Clients.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/Clients.java @@ -29,12 +29,19 @@ 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(); + boolean willFlag = options.getWillFlagSnapshot(); + int willQoS = options.getWillQoSSnapshot(); + boolean willRetain = options.getWillRetainSnapshot(); + + String id = String.format("%s@%s:%s<%s>-[%s]-ssl:%s-trustAll:%s-will:%s:%d:%s", + username, host, port, server, clientId, ssl, trustAll, willFlag, willQoS, willRetain); - 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 +56,12 @@ public static void clear() { public static class ClientHolder { private final MqttClientSession client; + private final MqttClientSessionOptions options; private final BroadcastProcessor 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))); } diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java index f466d23efe..0923040612 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java @@ -48,8 +48,8 @@ @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 topic of the will message", defaultValue = "") -@ConnectorAttribute(name = "will-payload", type = "string", direction = INCOMING_AND_OUTGOING, description = "Set the payload of the will message", defaultValue = "") +@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") @@ -68,6 +68,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"; diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttMessage.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttMessage.java index 4ebb7c1b7b..36e78b5d35 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttMessage.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttMessage.java @@ -3,6 +3,7 @@ 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 +37,21 @@ static MqttMessage of(String topic, T payload, MqttQoS qos, boolean retai return new SendingMqttMessage<>(payload, new SendingMqttMessageMetadata(topic, qos, retain)); } + static MqttMessage of(String topic, T payload, MqttQoS qos, boolean retain, MqttProperties properties) { + return new SendingMqttMessage<>(payload, new SendingMqttMessageMetadata(topic, qos, retain, properties)); + } + + static MqttMessage of(String topic, T payload, MqttProperties properties) { + return new SendingMqttMessage<>(payload, new SendingMqttMessageMetadata(topic, null, false, properties)); + } + + // TODO Should be removed? default MqttMessage withAck(Supplier> 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(); diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSink.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSink.java index dbe1489d6f..b63c8fcf04 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSink.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSink.java @@ -11,6 +11,7 @@ import org.eclipse.microprofile.reactive.messaging.Message; +import io.netty.handler.codec.mqtt.MqttProperties; import io.netty.handler.codec.mqtt.MqttQoS; import io.smallrye.mutiny.Uni; import io.smallrye.mutiny.vertx.AsyncResultUni; @@ -86,6 +87,7 @@ private Uni> send(Message msg) { final String actualTopicToBeUsed; final MqttQoS actualQoS; final boolean isRetain; + final MqttProperties properties; Optional metadata = msg.getMetadata(SendingMqttMessageMetadata.class); if (metadata.isPresent()) { @@ -93,10 +95,12 @@ private Uni> send(Message msg) { actualTopicToBeUsed = mm.getTopic() == null ? this.topic : mm.getTopic(); actualQoS = mm.getQosLevel() == null ? MqttQoS.valueOf(this.qos) : mm.getQosLevel(); isRetain = mm.isRetain(); + properties = mm.getProperties(); } else { actualTopicToBeUsed = this.topic; isRetain = this.retain; actualQoS = MqttQoS.valueOf(this.qos); + properties = MqttProperties.NO_PROPERTIES; } if (actualTopicToBeUsed == null) { @@ -106,7 +110,8 @@ private Uni> send(Message msg) { return AsyncResultUni . toUni(h -> client - .publish(actualTopicToBeUsed, convert(msg.getPayload()).getDelegate(), actualQoS, false, isRetain) + .publish(actualTopicToBeUsed, convert(msg.getPayload()).getDelegate(), actualQoS, false, isRetain, + properties) .onComplete(h)) .onItemOrFailure().transformToUni((s, f) -> { if (f != null) { diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSource.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSource.java index 448554bb78..d0aa7f5a23 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSource.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSource.java @@ -6,8 +6,9 @@ import java.util.concurrent.Flow; import java.util.concurrent.atomic.AtomicBoolean; -import jakarta.enterprise.inject.Instance; - +import io.netty.handler.codec.mqtt.MqttQoS; +import io.netty.handler.codec.mqtt.MqttSubscriptionOption; +import io.netty.handler.codec.mqtt.MqttSubscriptionOption.RetainedHandlingPolicy; import io.smallrye.mutiny.Uni; import io.smallrye.reactive.messaging.ClientCustomizer; import io.smallrye.reactive.messaging.health.HealthReport.HealthReportBuilder; @@ -15,12 +16,12 @@ import io.smallrye.reactive.messaging.mqtt.internal.MqttHelpers; import io.smallrye.reactive.messaging.mqtt.internal.MqttTopicHelper; import io.smallrye.reactive.messaging.mqtt.session.MqttClientSessionOptions; -import io.smallrye.reactive.messaging.mqtt.session.RequestedQoS; import io.smallrye.reactive.messaging.providers.helpers.ConfigUtils; import io.smallrye.reactive.messaging.providers.helpers.VertxContext; import io.vertx.core.impl.VertxInternal; import io.vertx.mutiny.core.Context; import io.vertx.mutiny.core.Vertx; +import jakarta.enterprise.inject.Instance; public class MqttSource { @@ -54,8 +55,21 @@ public MqttSource(Vertx vertx, MqttConnectorIncomingConfiguration config, final Context root = Context.newInstance(((VertxInternal) vertx.getDelegate()).createEventLoopContext()); holder = Clients.getHolder(vertx, options); holder.start().onSuccess(ignore -> started.set(true)); + + boolean noLocal = config.getNoLocal(); + boolean retainAsPublished = config.getRetainAsPublished(); + int retainHandlingValue = config.getRetainHandling(); + RetainedHandlingPolicy retainHandling; + try { + retainHandling = RetainedHandlingPolicy.valueOf(retainHandlingValue); + } catch (IllegalArgumentException e) { + throw ex.illegalArgumentInvalidRetainHandling(retainHandlingValue); + } + MqttSubscriptionOption subscriptionOption = new MqttSubscriptionOption( + MqttQoS.valueOf(qos), noLocal, retainAsPublished, retainHandling); + holder.getClient() - .subscribe(topic, RequestedQoS.valueOf(qos)) + .subscribe(topic, subscriptionOption) .onFailure(outcome -> log.info("Subscription failed!")) .onSuccess(outcome -> { log.info("Subscription success on topic " + topic + ", Max QoS " + outcome + "."); diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/ReceivingMqttMessageMetadata.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/ReceivingMqttMessageMetadata.java index b1de162ba8..da7b3ad99b 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/ReceivingMqttMessageMetadata.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/ReceivingMqttMessageMetadata.java @@ -1,5 +1,9 @@ package io.smallrye.reactive.messaging.mqtt; +import java.util.Collections; +import java.util.List; + +import io.netty.handler.codec.mqtt.MqttProperties; import io.netty.handler.codec.mqtt.MqttQoS; import io.vertx.mutiny.mqtt.messages.MqttPublishMessage; @@ -50,4 +54,68 @@ public boolean isDuplicate() { return message.isDup(); } + /** + * @return the MQTT 5.0 properties of the message, or {@code NO_PROPERTIES} for MQTT 3.1.1 + */ + public MqttProperties getProperties() { + return message.properties(); + } + + /** + * @return the User Properties from the MQTT 5.0 message, or empty list if not present + */ + public List getUserProperties() { + MqttProperties.MqttProperty prop = getProperties() + .getProperty(MqttProperties.MqttPropertyType.USER_PROPERTY.value()); + if (prop instanceof MqttProperties.UserProperties) { + return ((MqttProperties.UserProperties) prop).value(); + } + return Collections.emptyList(); + } + + /** + * @return the Content-Type property from the MQTT 5.0 message, or null if not present + */ + public String getContentType() { + MqttProperties.MqttProperty prop = getProperties() + .getProperty(MqttProperties.MqttPropertyType.CONTENT_TYPE.value()); + return prop != null ? (String) prop.value() : null; + } + + /** + * @return the Response Topic property from the MQTT 5.0 message, or null if not present + */ + public String getResponseTopic() { + MqttProperties.MqttProperty prop = getProperties() + .getProperty(MqttProperties.MqttPropertyType.RESPONSE_TOPIC.value()); + return prop != null ? (String) prop.value() : null; + } + + /** + * @return the Correlation Data property from the MQTT 5.0 message, or null if not present + */ + public byte[] getCorrelationData() { + MqttProperties.MqttProperty prop = getProperties() + .getProperty(MqttProperties.MqttPropertyType.CORRELATION_DATA.value()); + return prop != null ? (byte[]) prop.value() : null; + } + + /** + * @return the Message Expiry Interval from the MQTT 5.0 message (in seconds), or null if not present + */ + public Integer getMessageExpiryInterval() { + MqttProperties.MqttProperty prop = getProperties() + .getProperty(MqttProperties.MqttPropertyType.PUBLICATION_EXPIRY_INTERVAL.value()); + return prop != null ? (Integer) prop.value() : null; + } + + /** + * @return the Payload Format Indicator (0=binary, 1=UTF-8), or null if not present + */ + public Integer getPayloadFormatIndicator() { + MqttProperties.MqttProperty prop = getProperties() + .getProperty(MqttProperties.MqttPropertyType.PAYLOAD_FORMAT_INDICATOR.value()); + return prop != null ? (Integer) prop.value() : null; + } + } diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/SendingMqttMessageMetadata.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/SendingMqttMessageMetadata.java index 1cbd76cc18..6f19128868 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/SendingMqttMessageMetadata.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/SendingMqttMessageMetadata.java @@ -1,5 +1,6 @@ package io.smallrye.reactive.messaging.mqtt; +import io.netty.handler.codec.mqtt.MqttProperties; import io.netty.handler.codec.mqtt.MqttQoS; /** @@ -10,11 +11,17 @@ public final class SendingMqttMessageMetadata implements MqttMessageMetadata { private final String topic; private final MqttQoS qos; private final boolean isRetain; + private final MqttProperties properties; public SendingMqttMessageMetadata(String topic, MqttQoS qos, boolean isRetain) { + this(topic, qos, isRetain, MqttProperties.NO_PROPERTIES); + } + + public SendingMqttMessageMetadata(String topic, MqttQoS qos, boolean isRetain, MqttProperties properties) { this.topic = topic; this.qos = qos; this.isRetain = isRetain; + this.properties = properties != null ? properties : MqttProperties.NO_PROPERTIES; } @Override @@ -31,4 +38,8 @@ public MqttQoS getQosLevel() { public boolean isRetain() { return isRetain; } + + public MqttProperties getProperties() { + return properties; + } } diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java index bddbda520a..f59ec525d5 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java @@ -22,4 +22,7 @@ public interface MqttExceptions { @Message(id = 17002, value = "Cannot find a %s bean identified with %s") IllegalStateException illegalStateFindingBean(String className, String beanName); + @Message(id = 17003, value = "Invalid retain-handling value: %d. Must be 0, 1, or 2.") + IllegalArgumentException illegalArgumentInvalidRetainHandling(int value); + } diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java index c55b4c1717..d9a0f0067f 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java @@ -22,6 +22,7 @@ import io.vertx.core.net.PemTrustOptions; import io.vertx.core.net.PfxOptions; import io.vertx.core.net.TrustOptions; +import io.vertx.mqtt.MqttClientWillOptions; public class MqttHelpers { @@ -57,15 +58,25 @@ private static MqttClientSessionOptions createMqttClientOptions(MqttConnectorCom options.setTrustOptions(getTrustOptions(config)); options.setTrustAll(config.getTrustAll()); options.setUsername(config.getUsername().orElse(null)); - options.setWillQoS(config.getWillQos()); options.setWillFlag(config.getWillFlag()); - options.setWillRetain(config.getWillRetain()); - options.setWillTopic(config.getWillTopic()); - options.setWillMessageBytes(Buffer.buffer(config.getWillPayload())); options.setUnsubscribeOnDisconnect(config.getUnsubscribeOnDisconnection()); options.setMetricsName("mqtt|" + config.getChannel()); - config.config().getOptionalValue("version", Integer.class).ifPresent(v -> options.setVersion(v)); + options.setVersion(config.getVersion()); + config.getSessionExpiryInterval().ifPresent(options::setSessionExpireInterval); + config.getReceiveMaximum().ifPresent(options::setReceiveMaximum); + options.setTopicAliasMaximum(config.getTopicAliasMaximum()); + + if (config.getWillFlag()) { + options.setWillTopic(config.getWillTopic()); + options.setWillQoS(config.getWillQos()); + options.setWillRetain(config.getWillRetain()); + options.setWillMessageBytes(Buffer.buffer(config.getWillPayload())); + MqttClientWillOptions willOptions = options.getWillOptions(); + config.getWillContentType().ifPresent(willOptions::setContentType); + config.getWillResponseTopic().ifPresent(willOptions::setResponseTopic); + config.getWillDelayInterval().ifPresent(willOptions::setWillDelayInterval); + } return options; } @@ -208,28 +219,65 @@ static void merge(MqttClientSessionOptions custom, MqttConnectorCommonConfigurat custom.setTrustAll(config.getTrustAll()); } - if (isSetInChannelConfiguration("will-qos", config)) { - custom.setWillQoS(config.getWillQos()); + if (isSetInChannelConfiguration("unsubscribe-on-disconnection", config)) { + custom.setUnsubscribeOnDisconnect(config.getUnsubscribeOnDisconnection()); } - if (isSetInChannelConfiguration("will-flag", config)) { - custom.setWillFlag(config.getWillFlag()); + if (DEFAULT_METRICS_NAME.equals(custom.getMetricsName())) { + custom.setMetricsName("mqtt|" + config.getChannel()); } - if (isSetInChannelConfiguration("will-retain", config)) { - custom.setWillRetain(config.getWillRetain()); + if (isSetInChannelConfiguration("version", config)) { + custom.setVersion(config.getVersion()); } - if (isSetInChannelConfiguration("will-topic", config)) { - custom.setWillTopic(config.getWillTopic()); + + if (isSetInChannelConfiguration("session-expiry-interval", config)) { + config.getSessionExpiryInterval().ifPresent(custom::setSessionExpireInterval); } - if (isSetInChannelConfiguration("will-payload", config)) { - custom.setWillMessageBytes(Buffer.buffer(config.getWillPayload())); + + if (isSetInChannelConfiguration("receive-maximum", config)) { + config.getReceiveMaximum().ifPresent(custom::setReceiveMaximum); } - if (isSetInChannelConfiguration("unsubscribe-on-disconnection", config)) { - custom.setUnsubscribeOnDisconnect(config.getUnsubscribeOnDisconnection()); + + if (isSetInChannelConfiguration("topic-alias-maximum", config)) { + custom.setTopicAliasMaximum(config.getTopicAliasMaximum()); } - if (DEFAULT_METRICS_NAME.equals(custom.getMetricsName())) { - custom.setMetricsName("mqtt|" + config.getChannel()); + + if (config.getWillFlag()) { + + if (isSetInChannelConfiguration("will-qos", config)) { + custom.setWillQoS(config.getWillQos()); + } + + if (isSetInChannelConfiguration("will-flag", config)) { + custom.setWillFlag(config.getWillFlag()); + } + + if (isSetInChannelConfiguration("will-retain", config)) { + custom.setWillRetain(config.getWillRetain()); + } + + if (isSetInChannelConfiguration("will-topic", config)) { + custom.setWillTopic(config.getWillTopic()); + } + + if (isSetInChannelConfiguration("will-payload", config)) { + custom.setWillMessageBytes(Buffer.buffer(config.getWillPayload())); + } + + MqttClientWillOptions willOpts = custom.getWillOptions(); + if (isSetInChannelConfiguration("will-content-type", config)) { + config.getWillContentType().ifPresent(willOpts::setContentType); + } + + if (isSetInChannelConfiguration("will-response-topic", config)) { + config.getWillResponseTopic().ifPresent(willOpts::setResponseTopic); + } + + if (isSetInChannelConfiguration("will-delay-interval", config)) { + config.getWillDelayInterval().ifPresent(willOpts::setWillDelayInterval); + } + } } diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSession.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSession.java index 61c7327594..908e3b619d 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSession.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSession.java @@ -16,7 +16,9 @@ package io.smallrye.reactive.messaging.mqtt.session; +import io.netty.handler.codec.mqtt.MqttProperties; import io.netty.handler.codec.mqtt.MqttQoS; +import io.netty.handler.codec.mqtt.MqttSubscriptionOption; import io.smallrye.reactive.messaging.mqtt.session.impl.MqttClientSessionImpl; import io.vertx.codegen.annotations.Fluent; import io.vertx.codegen.annotations.VertxGen; @@ -137,6 +139,15 @@ default boolean isConnected() { */ Future subscribe(String topic, RequestedQoS qos); + /** + * Subscribes to a single topic with MQTT 5.0 subscription options. + * + * @param topic The topic to subscribe to. + * @param subscriptionOption The v5 subscription option, or null to use default options. + * @return a {@code Future} completed with the granted QoS + */ + Future subscribe(String topic, MqttSubscriptionOption subscriptionOption); + /** * Unsubscribe from receiving messages on given topic * @@ -175,6 +186,20 @@ default boolean isConnected() { */ Future publish(String topic, Buffer payload, MqttQoS qosLevel, boolean isDup, boolean isRetain); + /** + * Sends the PUBLISH message with MQTT 5.0 properties. + * + * @param topic topic on which the message is published + * @param payload message payload + * @param qosLevel QoS level + * @param isDup if the message is a duplicate + * @param isRetain if the message needs to be retained + * @param properties MQTT 5.0 properties; use {@link MqttProperties#NO_PROPERTIES} for none + * @return a {@code Future} completed after PUBLISH packet sent with packetid (not when QoS 0) + */ + Future publish(String topic, Buffer payload, MqttQoS qosLevel, boolean isDup, boolean isRetain, + MqttProperties properties); + /** * Sends the PUBLISH message to the remote MQTT server * diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java index 4905652a0f..26b188dd1b 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java @@ -31,6 +31,11 @@ public class MqttClientSessionOptions extends MqttClientOptions { private ReconnectDelayOptions reconnectDelay = DEFAULT_RECONNECT_DELAY; private boolean unsubscribeOnDisconnect = false; + // Snapshots of WILL settings for easy access without reflection + private boolean willFlagSnapshot = false; + private int willQoSSnapshot = 0; + private boolean willRetainSnapshot = false; + /** * Default constructor */ @@ -49,6 +54,9 @@ public MqttClientSessionOptions(MqttClientSessionOptions other) { this.port = other.port; this.serverName = other.serverName; this.reconnectDelay = other.reconnectDelay.copy(); + this.willFlagSnapshot = other.willFlagSnapshot; + this.willQoSSnapshot = other.willQoSSnapshot; + this.willRetainSnapshot = other.willRetainSnapshot; } public int getPort() { @@ -141,21 +149,36 @@ public MqttClientSessionOptions setCleanSession(boolean cleanSession) { @Override public MqttClientSessionOptions setWillFlag(boolean willFlag) { super.setWillFlag(willFlag); + this.willFlagSnapshot = willFlag; return this; } @Override public MqttClientSessionOptions setWillQoS(int willQoS) { super.setWillQoS(willQoS); + this.willQoSSnapshot = willQoS; return this; } @Override public MqttClientSessionOptions setWillRetain(boolean willRetain) { super.setWillRetain(willRetain); + this.willRetainSnapshot = willRetain; return this; } + public boolean getWillFlagSnapshot() { + return willFlagSnapshot; + } + + public int getWillQoSSnapshot() { + return willQoSSnapshot; + } + + public boolean getWillRetainSnapshot() { + return willRetainSnapshot; + } + @Override public MqttClientSessionOptions setKeepAliveInterval(int keepAliveInterval) { super.setKeepAliveInterval(keepAliveInterval); @@ -501,4 +524,8 @@ public ClientOptionsBase setTcpUserTimeout(int tcpUserTimeout) { super.setTcpUserTimeout(tcpUserTimeout); return this; } + + // Note: setVersion, setSessionExpireInterval, setReceiveMaximum, setTopicAliasMaximum + // are inherited from MqttClientOptions and return void. We don't override them + // to avoid return type incompatibility issues. } diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/impl/MqttClientSessionImpl.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/impl/MqttClientSessionImpl.java index 0f1f038d21..3631f11b0d 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/impl/MqttClientSessionImpl.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/impl/MqttClientSessionImpl.java @@ -11,7 +11,9 @@ import java.util.concurrent.RejectedExecutionException; import java.util.stream.Collectors; +import io.netty.handler.codec.mqtt.MqttProperties; import io.netty.handler.codec.mqtt.MqttQoS; +import io.netty.handler.codec.mqtt.MqttSubscriptionOption; import io.smallrye.reactive.messaging.mqtt.session.MqttClientSession; import io.smallrye.reactive.messaging.mqtt.session.MqttClientSessionOptions; import io.smallrye.reactive.messaging.mqtt.session.ReconnectDelayProvider; @@ -43,9 +45,9 @@ public class MqttClientSessionImpl implements MqttClientSession { private final MqttClientSessionOptions options; // record the subscriptions - private final Map subscriptions = new HashMap<>(); + private final Map subscriptions = new HashMap<>(); // record the pending subscribes - private final Map> pendingSubscribes = new HashMap<>(); + private final Map> pendingSubscribes = new HashMap<>(); // record the pending unsubscribes private final Map> pendingUnsubscribes = new HashMap<>(); // the provider for the reconnect delay @@ -119,7 +121,17 @@ public SubscriptionState getSubscriptionState(String topicFilter) { @Override public Future subscribe(String topic, RequestedQoS qos) { Promise result = Promise.promise(); - this.vertx.runOnContext(x -> doSubscribe(topic, qos, result)); + this.vertx.runOnContext(x -> doSubscribe(topic, qos, null, result)); + return result.future(); + } + + @Override + public Future subscribe(String topic, MqttSubscriptionOption subscriptionOption) { + Promise result = Promise.promise(); + this.vertx.runOnContext(x -> { + RequestedQoS requestedQoS = RequestedQoS.valueOf(subscriptionOption.qos().value()); + doSubscribe(topic, requestedQoS, subscriptionOption, result); + }); return result.future(); } @@ -498,7 +510,7 @@ private void connectCompleted(AsyncResult result) { // If the session is present on broker, I mark all subscription to SUBSC log.debug("Session present on broker, subscriptions request not sent. " + "Be sure that the subscriptions on the broker side are the same that this client needs."); - this.subscriptions.forEach((t, q) -> notifySubscriptionState(t, SubscriptionState.SUBSCRIBED, q.toInteger())); + this.subscriptions.forEach((t, e) -> notifySubscriptionState(t, SubscriptionState.SUBSCRIBED, e.qos.toInteger())); } } @@ -582,26 +594,31 @@ private void serverPublished(MqttPublishMessage message) { /** * Perform subscribing. * - * @param topic The topics to subscribe to. + * @param topic The topic to subscribe to. + * @param qos The QoS to request. + * @param option MQTT 5.0 subscription option, or null for default. + * @param handler Callback for subscription result. */ - private void doSubscribe(String topic, RequestedQoS qos, Handler> handler) { + private void doSubscribe(String topic, RequestedQoS qos, MqttSubscriptionOption option, + Handler> handler) { if (log.isDebugEnabled()) { log.debug(String.format("Request to subscribe to: %s / %s", topic, qos)); } - RequestedQoS current = this.subscriptions.get(topic); + SubscriptionEntry current = this.subscriptions.get(topic); if (current != null) { if (log.isDebugEnabled()) { - log.debug("Already subscribed with: " + current); + log.debug("Already subscribed with qos " + current.qos); } if (handler != null) { - handler.handle(Future.succeededFuture(current.toInteger())); + handler.handle(Future.succeededFuture(current.qos.toInteger())); } return; } - this.subscriptions.put(topic, qos); + SubscriptionEntry entry = new SubscriptionEntry(qos, option); + this.subscriptions.put(topic, entry); if (handler != null) { this.notifySubscribed.computeIfAbsent(topic, x -> new LinkedList<>()) @@ -611,7 +628,7 @@ private void doSubscribe(String topic, RequestedQoS qos, Handler(Collections.singletonMap(topic, qos))); + requestSubscribe(new LinkedHashMap<>(Collections.singletonMap(topic, entry))); } /** @@ -640,9 +657,9 @@ private void doUnsubscribe(String topic, Handler> handler) { /** * Request to subscribe from the server. * - * @param topics The topics to subscribe to, including the requested QoS. + * @param topics The topics to subscribe to, including the requested QoS and subscription options. */ - private void requestSubscribe(LinkedHashMap topics) { + private void requestSubscribe(LinkedHashMap topics) { if (topics.isEmpty() || this.client == null || !this.client.isConnected()) { // nothing to do return; @@ -652,11 +669,16 @@ private void requestSubscribe(LinkedHashMap topics) { log.debug("Request Subscribe to: " + topics); } - this.client - .subscribe(topics.entrySet() - .stream().collect(Collectors.toMap( - Map.Entry::getKey, - e -> e.getValue().toInteger()))) + // For MQTT v5, we would use MqttTopicSubscription with subscription options, + // but for compatibility with vertx-mqtt 4.5.27, we use Map-based subscribe which works for v4 and v5. + // The subscription options are stored in our SubscriptionEntry and will be + // automatically re-applied on reconnect via the subscriptions map. + java.util.Map topicsMap = topics.entrySet().stream() + .collect(Collectors.toMap( + Map.Entry::getKey, + e -> e.getValue().qos.toInteger())); + + this.client.subscribe(topicsMap, MqttProperties.NO_PROPERTIES) .onComplete(result -> subscribeSent(result, topics)); } @@ -684,7 +706,7 @@ private void requestUnsubscribe(List topics) { * * @param result The result of sending the request, contains the packet id. */ - private void subscribeSent(AsyncResult result, LinkedHashMap topics) { + private void subscribeSent(AsyncResult result, LinkedHashMap topics) { if (result.failed() || result.result() == null) { // failed for (String topic : topics.keySet()) { @@ -718,7 +740,7 @@ private void unsubscribeSent(AsyncResult result, List topics) { * @param ack The acknowledge message. */ private void subscribeCompleted(MqttSubAckMessage ack) { - LinkedHashMap request = this.pendingSubscribes.remove(ack.messageId()); + LinkedHashMap request = this.pendingSubscribes.remove(ack.messageId()); if (request == null) { closeConnection(String.format("Unexpected subscription ack response - messageId: %s", ack.messageId())); return; @@ -753,17 +775,22 @@ private void unsubscribeCompleted(Integer messageId) { @Override public Future publish(String topic, Buffer payload, MqttQoS qosLevel, boolean isDup, boolean isRetain) { + return publish(topic, payload, qosLevel, isDup, isRetain, MqttProperties.NO_PROPERTIES); + } + + @Override + public Future publish(String topic, Buffer payload, MqttQoS qosLevel, boolean isDup, boolean isRetain, + MqttProperties properties) { Promise future = Promise.promise(); - this.vertx - .runOnContext(x -> doPublish(topic, payload, qosLevel, isDup, isRetain) - .onComplete(future)); + this.vertx.runOnContext( + x -> doPublish(topic, payload, qosLevel, isDup, isRetain, properties).onComplete(future)); return future.future(); } - private Future doPublish(String topic, Buffer payload, MqttQoS qosLevel, boolean isDup, boolean isRetain) { + private Future doPublish(String topic, Buffer payload, MqttQoS qosLevel, boolean isDup, boolean isRetain, + MqttProperties properties) { if (this.client != null && this.client.isConnected()) { - // not checking for isConnected might throw a NPE from inside the client - return this.client.publish(topic, payload, qosLevel, isDup, isRetain); + return this.client.publish(topic, payload, qosLevel, isDup, isRetain, properties); } else { return Future.failedFuture("Session is not connected"); } diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/impl/SubscriptionEntry.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/impl/SubscriptionEntry.java new file mode 100644 index 0000000000..67f82128b6 --- /dev/null +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/impl/SubscriptionEntry.java @@ -0,0 +1,18 @@ +package io.smallrye.reactive.messaging.mqtt.session.impl; + +import io.netty.handler.codec.mqtt.MqttQoS; +import io.netty.handler.codec.mqtt.MqttSubscriptionOption; +import io.smallrye.reactive.messaging.mqtt.session.RequestedQoS; + +final class SubscriptionEntry { + + final RequestedQoS qos; + final MqttSubscriptionOption option; + + SubscriptionEntry(RequestedQoS qos, MqttSubscriptionOption option) { + this.qos = qos; + this.option = option != null ? option + : MqttSubscriptionOption.onlyFromQos(MqttQoS.valueOf(qos.toInteger())); + } + +} \ No newline at end of file From 26089d9397f1db284c1fd997da063490c81af488 Mon Sep 17 00:00:00 2001 From: Domenico Briganti Date: Sun, 3 May 2026 15:01:27 +0200 Subject: [PATCH 05/11] Add validation for will-topic and will-payload in MqttHelpers and new exception message --- .../java/io/smallrye/reactive/messaging/mqtt/MqttSource.java | 3 ++- .../smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java | 3 +++ .../smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java | 3 +++ 3 files changed, 8 insertions(+), 1 deletion(-) diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSource.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSource.java index d0aa7f5a23..7ee71c5995 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSource.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttSource.java @@ -6,6 +6,8 @@ import java.util.concurrent.Flow; import java.util.concurrent.atomic.AtomicBoolean; +import jakarta.enterprise.inject.Instance; + import io.netty.handler.codec.mqtt.MqttQoS; import io.netty.handler.codec.mqtt.MqttSubscriptionOption; import io.netty.handler.codec.mqtt.MqttSubscriptionOption.RetainedHandlingPolicy; @@ -21,7 +23,6 @@ import io.vertx.core.impl.VertxInternal; import io.vertx.mutiny.core.Context; import io.vertx.mutiny.core.Vertx; -import jakarta.enterprise.inject.Instance; public class MqttSource { diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java index f59ec525d5..b76d7143a5 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java @@ -25,4 +25,7 @@ public interface MqttExceptions { @Message(id = 17003, value = "Invalid retain-handling value: %d. Must be 0, 1, or 2.") IllegalArgumentException illegalArgumentInvalidRetainHandling(int value); + @Message(id = 17004, value = "When 'will-flag' is enabled on channel '%s', both 'will-topic' and 'will-payload' must be set") + IllegalArgumentException illegalArgumentMissingWillTopicOrPayload(String channel); + } diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java index d9a0f0067f..8934072a1f 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java @@ -68,6 +68,9 @@ private static MqttClientSessionOptions createMqttClientOptions(MqttConnectorCom options.setTopicAliasMaximum(config.getTopicAliasMaximum()); if (config.getWillFlag()) { + if (config.getWillTopic().isEmpty() || config.getWillPayload().isEmpty()) { + throw ex.illegalArgumentMissingWillTopicOrPayload(config.getChannel()); + } options.setWillTopic(config.getWillTopic()); options.setWillQoS(config.getWillQos()); options.setWillRetain(config.getWillRetain()); From 82d626d249b39bd286968ff414ad3d40884d1c14 Mon Sep 17 00:00:00 2001 From: Domenico Briganti Date: Sun, 3 May 2026 16:27:32 +0200 Subject: [PATCH 06/11] fix typo --- .../reactive/messaging/providers/i18n/ProviderExceptions.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/i18n/ProviderExceptions.java b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/i18n/ProviderExceptions.java index 2f2629c01b..2347be4816 100644 --- a/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/i18n/ProviderExceptions.java +++ b/smallrye-reactive-messaging-provider/src/main/java/io/smallrye/reactive/messaging/providers/i18n/ProviderExceptions.java @@ -227,7 +227,7 @@ IllegalArgumentException illegalArgumentForWorkerConfigKey(String annotation, St @Message(id = 68, value = "The operation %s has returned null") NullPointerException nullPointerOnInvokeBlocking(String methodAsString); - @Message(id = 71, value = "Invalid channel configuration - the `connector` attribute must be set for channel `%s`") + @Message(id = 71, value = "Invalid channel configuration - the `connector` attribute must be set for channel `%s`") IllegalArgumentException illegalArgumentChannelConnectorConfiguration(String name); @Message(id = 72, value = "Unknown connector for `%s`.") From 60700861aa6c0fbf44d0ca56bc1281c4ad2e0764 Mon Sep 17 00:00:00 2001 From: Domenico Briganti Date: Sun, 3 May 2026 17:41:52 +0200 Subject: [PATCH 07/11] Add response message handling to MqttMessage interface and update exceptions --- .../reactive/messaging/mqtt/MqttMessage.java | 33 +++++++ .../messaging/mqtt/i18n/MqttExceptions.java | 3 + .../messaging/mqtt/MqttMessageTest.java | 94 +++++++++++++++++++ 3 files changed, 130 insertions(+) diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttMessage.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttMessage.java index 36e78b5d35..58e83e99f3 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttMessage.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttMessage.java @@ -1,5 +1,7 @@ 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; @@ -45,6 +47,25 @@ static MqttMessage of(String topic, T payload, MqttProperties properties) return new SendingMqttMessage<>(payload, new SendingMqttMessageMetadata(topic, null, false, properties)); } + static MqttMessage ofResponse(MqttMessage inputMessage, T payload, MqttQoS qos) { + return ofResponse(inputMessage, payload, qos, new MqttProperties()); + } + + static MqttMessage 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 withAck(Supplier> ack) { MqttProperties props = getMetadata(SendingMqttMessageMetadata.class) @@ -63,4 +84,16 @@ default MqttMessage withAck(Supplier> 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); + } } diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java index b76d7143a5..6decd22bdc 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/i18n/MqttExceptions.java @@ -28,4 +28,7 @@ public interface MqttExceptions { @Message(id = 17004, value = "When 'will-flag' is enabled on channel '%s', both 'will-topic' and 'will-payload' must be set") IllegalArgumentException illegalArgumentMissingWillTopicOrPayload(String channel); + @Message(id = 17005, value = "Cannot build a response message: the input message has no MQTT 5 'Response Topic' property") + IllegalArgumentException illegalArgumentMissingResponseTopic(); + } diff --git a/smallrye-reactive-messaging-mqtt/src/test/java/io/smallrye/reactive/messaging/mqtt/MqttMessageTest.java b/smallrye-reactive-messaging-mqtt/src/test/java/io/smallrye/reactive/messaging/mqtt/MqttMessageTest.java index ee78623dcb..b657ee394e 100644 --- a/smallrye-reactive-messaging-mqtt/src/test/java/io/smallrye/reactive/messaging/mqtt/MqttMessageTest.java +++ b/smallrye-reactive-messaging-mqtt/src/test/java/io/smallrye/reactive/messaging/mqtt/MqttMessageTest.java @@ -1,9 +1,11 @@ package io.smallrye.reactive.messaging.mqtt; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import org.junit.jupiter.api.Test; +import io.netty.handler.codec.mqtt.MqttProperties; import io.netty.handler.codec.mqtt.MqttQoS; public class MqttMessageTest { @@ -83,4 +85,96 @@ void testOfMetadata() { assertThat(message.isDuplicate()).isFalse(); assertThat(message.isRetain()).isTrue(); } + + @Test + void testOfResponsePropagatesResponseTopicQosAndCorrelationData() { + byte[] correlation = new byte[] { 1, 2, 3 }; + MqttMessage incoming = fakeIncoming("response/topic", correlation); + + MqttMessage reply = MqttMessage.ofResponse(incoming, "pong", MqttQoS.AT_LEAST_ONCE); + + assertThat(reply.getPayload()).isEqualTo("pong"); + assertThat(reply.getTopic()).isEqualTo("response/topic"); + assertThat(reply.getQosLevel()).isEqualTo(MqttQoS.AT_LEAST_ONCE); + + SendingMqttMessageMetadata md = reply.getMetadata(SendingMqttMessageMetadata.class).orElseThrow(); + MqttProperties.MqttProperty cd = md.getProperties() + .getProperty(MqttProperties.MqttPropertyType.CORRELATION_DATA.value()); + assertThat(cd).isNotNull(); + assertThat((byte[]) cd.value()).containsExactly(1, 2, 3); + } + + @Test + void testOfResponseOmitsCorrelationDataWhenAbsent() { + MqttMessage incoming = fakeIncoming("response/topic", null); + + MqttMessage reply = MqttMessage.ofResponse(incoming, "pong", MqttQoS.AT_MOST_ONCE); + + SendingMqttMessageMetadata md = reply.getMetadata(SendingMqttMessageMetadata.class).orElseThrow(); + assertThat(md.getProperties() + .getProperty(MqttProperties.MqttPropertyType.CORRELATION_DATA.value())).isNull(); + } + + @Test + void testOfResponseThrowsWhenResponseTopicMissing() { + MqttMessage incoming = fakeIncoming(null, null); + + assertThatThrownBy(() -> MqttMessage.ofResponse(incoming, "pong", MqttQoS.AT_MOST_ONCE)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Response Topic"); + } + + @Test + void testOfResponseDoesNotMutateCallerProperties() { + MqttMessage incoming = fakeIncoming("response/topic", new byte[] { 9 }); + MqttProperties callerProps = new MqttProperties(); + + MqttMessage.ofResponse(incoming, "pong", MqttQoS.AT_MOST_ONCE, callerProps); + + assertThat(callerProps.listAll()).isEmpty(); + } + + private static MqttMessage fakeIncoming(String responseTopic, byte[] correlationData) { + return new MqttMessage<>() { + @Override + public byte[] getPayload() { + return new byte[0]; + } + + @Override + public int getMessageId() { + return -1; + } + + @Override + public MqttQoS getQosLevel() { + return MqttQoS.AT_MOST_ONCE; + } + + @Override + public boolean isDuplicate() { + return false; + } + + @Override + public boolean isRetain() { + return false; + } + + @Override + public String getTopic() { + return "request/topic"; + } + + @Override + public String getResponseTopic() { + return responseTopic; + } + + @Override + public byte[] getCorrelationData() { + return correlationData; + } + }; + } } From 9823b5ff6e9844656fa8fb962e79261303ef0fd1 Mon Sep 17 00:00:00 2001 From: Domenico Briganti Date: Fri, 8 May 2026 09:28:11 +0200 Subject: [PATCH 08/11] Remove will-flag handling from MqttConnector and MqttHelpers, update version in pom.xml --- smallrye-reactive-messaging-mqtt/pom.xml | 2 +- .../messaging/mqtt/MqttConnector.java | 1 - .../messaging/mqtt/internal/MqttHelpers.java | 54 ++++++++----------- .../session/MqttClientSessionOptions.java | 7 --- 4 files changed, 24 insertions(+), 40 deletions(-) diff --git a/smallrye-reactive-messaging-mqtt/pom.xml b/smallrye-reactive-messaging-mqtt/pom.xml index e8816b228a..cfa53f6466 100644 --- a/smallrye-reactive-messaging-mqtt/pom.xml +++ b/smallrye-reactive-messaging-mqtt/pom.xml @@ -17,7 +17,7 @@ io.vertx vertx-mqtt - 4.5.27-SNAPSHOT + 4.5.28-SNAPSHOT diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java index 0923040612..5ec45771a4 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/MqttConnector.java @@ -45,7 +45,6 @@ @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 = "") diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java index 8934072a1f..5f237f1246 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/internal/MqttHelpers.java @@ -58,7 +58,6 @@ private static MqttClientSessionOptions createMqttClientOptions(MqttConnectorCom options.setTrustOptions(getTrustOptions(config)); options.setTrustAll(config.getTrustAll()); options.setUsername(config.getUsername().orElse(null)); - options.setWillFlag(config.getWillFlag()); options.setUnsubscribeOnDisconnect(config.getUnsubscribeOnDisconnection()); options.setMetricsName("mqtt|" + config.getChannel()); @@ -67,7 +66,7 @@ private static MqttClientSessionOptions createMqttClientOptions(MqttConnectorCom config.getReceiveMaximum().ifPresent(options::setReceiveMaximum); options.setTopicAliasMaximum(config.getTopicAliasMaximum()); - if (config.getWillFlag()) { + if (!config.getWillTopic().isEmpty() || !config.getWillPayload().isEmpty()) { if (config.getWillTopic().isEmpty() || config.getWillPayload().isEmpty()) { throw ex.illegalArgumentMissingWillTopicOrPayload(config.getChannel()); } @@ -246,42 +245,35 @@ static void merge(MqttClientSessionOptions custom, MqttConnectorCommonConfigurat custom.setTopicAliasMaximum(config.getTopicAliasMaximum()); } - if (config.getWillFlag()) { - - if (isSetInChannelConfiguration("will-qos", config)) { - custom.setWillQoS(config.getWillQos()); - } - - if (isSetInChannelConfiguration("will-flag", config)) { - custom.setWillFlag(config.getWillFlag()); - } - - if (isSetInChannelConfiguration("will-retain", config)) { - custom.setWillRetain(config.getWillRetain()); - } + if (isSetInChannelConfiguration("will-qos", config)) { + custom.setWillQoS(config.getWillQos()); + } - if (isSetInChannelConfiguration("will-topic", config)) { - custom.setWillTopic(config.getWillTopic()); - } + if (isSetInChannelConfiguration("will-retain", config)) { + custom.setWillRetain(config.getWillRetain()); + } - if (isSetInChannelConfiguration("will-payload", config)) { - custom.setWillMessageBytes(Buffer.buffer(config.getWillPayload())); - } + if (isSetInChannelConfiguration("will-topic", config)) { + custom.setWillTopic(config.getWillTopic()); + } - MqttClientWillOptions willOpts = custom.getWillOptions(); - if (isSetInChannelConfiguration("will-content-type", config)) { - config.getWillContentType().ifPresent(willOpts::setContentType); - } + if (isSetInChannelConfiguration("will-payload", config)) { + custom.setWillMessageBytes(Buffer.buffer(config.getWillPayload())); + } - if (isSetInChannelConfiguration("will-response-topic", config)) { - config.getWillResponseTopic().ifPresent(willOpts::setResponseTopic); - } + MqttClientWillOptions willOpts = custom.getWillOptions(); + if (isSetInChannelConfiguration("will-content-type", config)) { + config.getWillContentType().ifPresent(willOpts::setContentType); + } - if (isSetInChannelConfiguration("will-delay-interval", config)) { - config.getWillDelayInterval().ifPresent(willOpts::setWillDelayInterval); - } + if (isSetInChannelConfiguration("will-response-topic", config)) { + config.getWillResponseTopic().ifPresent(willOpts::setResponseTopic); + } + if (isSetInChannelConfiguration("will-delay-interval", config)) { + config.getWillDelayInterval().ifPresent(willOpts::setWillDelayInterval); } + } /** diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java index 26b188dd1b..bdb509e5fd 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java @@ -146,13 +146,6 @@ public MqttClientSessionOptions setCleanSession(boolean cleanSession) { return this; } - @Override - public MqttClientSessionOptions setWillFlag(boolean willFlag) { - super.setWillFlag(willFlag); - this.willFlagSnapshot = willFlag; - return this; - } - @Override public MqttClientSessionOptions setWillQoS(int willQoS) { super.setWillQoS(willQoS); From bd7cf036cf94e600bd04f92280167e3955afa0cc Mon Sep 17 00:00:00 2001 From: Domenico Briganti Date: Tue, 12 May 2026 10:38:30 +0200 Subject: [PATCH 09/11] Add MQTT 5.0 support and request/response handling in documentation and code --- documentation/src/main/docs/mqtt/mqtt.md | 33 +++++++++ .../main/docs/mqtt/receiving-mqtt-messages.md | 58 ++++++++++++++- .../docs/mqtt/sending-messages-to-mqtt.md | 71 ++++++++++++++++++- .../outbound/MqttRequestResponseHandler.java | 28 ++++++++ 4 files changed, 188 insertions(+), 2 deletions(-) create mode 100644 documentation/src/main/java/mqtt/outbound/MqttRequestResponseHandler.java diff --git a/documentation/src/main/docs/mqtt/mqtt.md b/documentation/src/main/docs/mqtt/mqtt.md index b75f991f49..c314441c08 100644 --- a/documentation/src/main/docs/mqtt/mqtt.md +++ b/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. + diff --git a/documentation/src/main/docs/mqtt/receiving-mqtt-messages.md b/documentation/src/main/docs/mqtt/receiving-mqtt-messages.md index 19bbfaa06d..1aa20d441c 100644 --- a/documentation/src/main/docs/mqtt/receiving-mqtt-messages.md +++ b/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`) | +| `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 diff --git a/documentation/src/main/docs/mqtt/sending-messages-to-mqtt.md b/documentation/src/main/docs/mqtt/sending-messages-to-mqtt.md index 9eb0ae1335..cd87b917e1 100644 --- a/documentation/src/main/docs/mqtt/sending-messages-to-mqtt.md +++ b/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 diff --git a/documentation/src/main/java/mqtt/outbound/MqttRequestResponseHandler.java b/documentation/src/main/java/mqtt/outbound/MqttRequestResponseHandler.java new file mode 100644 index 0000000000..9498d940ce --- /dev/null +++ b/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 handle(MqttMessage 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); + } +} From c9e74be8a92331c62c6930b468ceee0c25fcc252 Mon Sep 17 00:00:00 2001 From: Domenico Briganti Date: Tue, 12 May 2026 15:16:31 +0200 Subject: [PATCH 10/11] Refactor will-flag handling in MqttClientSessionOptions and update client ID format in Clients --- .../reactive/messaging/mqtt/Clients.java | 11 +++++----- .../session/MqttClientSessionOptions.java | 22 ------------------- 2 files changed, 6 insertions(+), 27 deletions(-) diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/Clients.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/Clients.java index c990060048..0a463a5241 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/Clients.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/Clients.java @@ -31,12 +31,13 @@ static ClientHolder getHolder(Vertx vertx, MqttClientSessionOptions options) { String username = options.getUsername(); boolean ssl = options.isSsl(); boolean trustAll = options.isTrustAll(); - boolean willFlag = options.getWillFlagSnapshot(); - int willQoS = options.getWillQoSSnapshot(); - boolean willRetain = options.getWillRetainSnapshot(); + 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", - username, host, port, server, clientId, ssl, trustAll, willFlag, willQoS, willRetain); + 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); return clients.computeIfAbsent(id, key -> { log.infof("Create MQTT Client for %s", id); diff --git a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java index bdb509e5fd..4a0b26d0ae 100644 --- a/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java +++ b/smallrye-reactive-messaging-mqtt/src/main/java/io/smallrye/reactive/messaging/mqtt/session/MqttClientSessionOptions.java @@ -31,11 +31,6 @@ public class MqttClientSessionOptions extends MqttClientOptions { private ReconnectDelayOptions reconnectDelay = DEFAULT_RECONNECT_DELAY; private boolean unsubscribeOnDisconnect = false; - // Snapshots of WILL settings for easy access without reflection - private boolean willFlagSnapshot = false; - private int willQoSSnapshot = 0; - private boolean willRetainSnapshot = false; - /** * Default constructor */ @@ -54,9 +49,6 @@ public MqttClientSessionOptions(MqttClientSessionOptions other) { this.port = other.port; this.serverName = other.serverName; this.reconnectDelay = other.reconnectDelay.copy(); - this.willFlagSnapshot = other.willFlagSnapshot; - this.willQoSSnapshot = other.willQoSSnapshot; - this.willRetainSnapshot = other.willRetainSnapshot; } public int getPort() { @@ -149,29 +141,15 @@ public MqttClientSessionOptions setCleanSession(boolean cleanSession) { @Override public MqttClientSessionOptions setWillQoS(int willQoS) { super.setWillQoS(willQoS); - this.willQoSSnapshot = willQoS; return this; } @Override public MqttClientSessionOptions setWillRetain(boolean willRetain) { super.setWillRetain(willRetain); - this.willRetainSnapshot = willRetain; return this; } - public boolean getWillFlagSnapshot() { - return willFlagSnapshot; - } - - public int getWillQoSSnapshot() { - return willQoSSnapshot; - } - - public boolean getWillRetainSnapshot() { - return willRetainSnapshot; - } - @Override public MqttClientSessionOptions setKeepAliveInterval(int keepAliveInterval) { super.setKeepAliveInterval(keepAliveInterval); From ab6fff7163b914e548116b328d770c3716d81f8d Mon Sep 17 00:00:00 2001 From: Domenico Briganti Date: Tue, 14 Jul 2026 18:21:23 +0200 Subject: [PATCH 11/11] Update vertx mqtt lib --- smallrye-reactive-messaging-mqtt/pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/smallrye-reactive-messaging-mqtt/pom.xml b/smallrye-reactive-messaging-mqtt/pom.xml index cfa53f6466..2fb0ae7697 100644 --- a/smallrye-reactive-messaging-mqtt/pom.xml +++ b/smallrye-reactive-messaging-mqtt/pom.xml @@ -17,7 +17,7 @@ io.vertx vertx-mqtt - 4.5.28-SNAPSHOT + 4.5.29-SNAPSHOT