diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java index b589da35fc..2116c3ed3e 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2023 the original author or authors. + * Copyright 2021-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package org.springframework.integration.mqtt.inbound; import java.util.Arrays; +import java.util.List; import java.util.Map; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -56,6 +57,8 @@ import org.springframework.messaging.MessagingException; import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.messaging.support.GenericMessage; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; +import org.springframework.util.ObjectUtils; /** * The {@link AbstractMqttMessageDrivenChannelAdapter} implementation for MQTT v5. @@ -83,10 +86,12 @@ public class Mqttv5PahoMessageDrivenChannelAdapter extends AbstractMqttMessageDrivenChannelAdapter implements MqttCallback, MqttComponent { - private final Lock lock = new ReentrantLock(); + private final Lock lock = new ReentrantLock(); private final MqttConnectionOptions connectionOptions; + private List subscriptions; + private IMqttAsyncClient mqttClient; @Nullable @@ -100,6 +105,18 @@ public class Mqttv5PahoMessageDrivenChannelAdapter private volatile boolean readyToSubscribeOnStart; + /** + * Create an instance based on the MQTT url, client id and subscriptions. + * @param url the MQTT url to connect. + * @param clientId the unique client id. + * @param mqttSubscriptions the MQTT subscriptions. + * @since 6.3 + */ + public Mqttv5PahoMessageDrivenChannelAdapter(String url, String clientId, MqttSubscription... mqttSubscriptions) { + this(url, clientId, Arrays.stream(mqttSubscriptions).map(MqttSubscription::getTopic).toArray(String[]::new)); + this.subscriptions = Arrays.asList(mqttSubscriptions); + } + public Mqttv5PahoMessageDrivenChannelAdapter(String url, String clientId, String... topic) { super(url, clientId, topic); Assert.hasText(url, "'url' cannot be null or empty"); @@ -108,6 +125,21 @@ public class Mqttv5PahoMessageDrivenChannelAdapter this.connectionOptions.setAutomaticReconnect(true); } + /** + * Create an instance based on the MQTT connection options, client id and subscriptions. + * @param connectionOptions the MQTT connection options. + * @param clientId the unique client id. + * @param mqttSubscriptions the MQTT subscriptions. + * @since 6.3 + */ + public Mqttv5PahoMessageDrivenChannelAdapter(MqttConnectionOptions connectionOptions, String clientId, + MqttSubscription... mqttSubscriptions) { + + this(connectionOptions, clientId, + Arrays.stream(mqttSubscriptions).map(MqttSubscription::getTopic).toArray(String[]::new)); + this.subscriptions = List.of(mqttSubscriptions); + } + public Mqttv5PahoMessageDrivenChannelAdapter(MqttConnectionOptions connectionOptions, String clientId, String... topic) { @@ -120,6 +152,19 @@ public class Mqttv5PahoMessageDrivenChannelAdapter } } + /** + * Create an instance based on the client manager and subscriptions. + * @param clientManager The client manager. + * @param mqttSubscriptions the MQTT subscriptions. + * @since 6.3 + */ + public Mqttv5PahoMessageDrivenChannelAdapter(ClientManager clientManager, + MqttSubscription... mqttSubscriptions) { + + this(clientManager, Arrays.stream(mqttSubscriptions).map(MqttSubscription::getTopic).toArray(String[]::new)); + this.subscriptions = List.of(mqttSubscriptions); + } + /** * Use this constructor when you need to use a single {@link ClientManager} * (for instance, to reuse an MQTT connection). @@ -270,13 +315,23 @@ public class Mqttv5PahoMessageDrivenChannelAdapter } } + @Override + public void setQos(int... qos) { + Assert.isNull(this.subscriptions, "The 'qos' must be provided with the 'MqttSubscription'."); + super.setQos(qos); + } + @Override public void addTopic(String topic, int qos) { this.topicLock.lock(); try { super.addTopic(topic, qos); + MqttSubscription subscription = new MqttSubscription(topic, qos); + if (this.subscriptions != null) { + this.subscriptions.add(subscription); + } if (this.mqttClient != null && this.mqttClient.isConnected()) { - this.mqttClient.subscribe(new MqttSubscription(topic, qos), this::messageArrived) + this.mqttClient.subscribe(subscription, this::messageArrived) .waitForCompletion(getCompletionTimeout()); } } @@ -296,6 +351,9 @@ public class Mqttv5PahoMessageDrivenChannelAdapter this.mqttClient.unsubscribe(topic).waitForCompletion(getCompletionTimeout()); } super.removeTopic(topic); + if (!CollectionUtils.isEmpty(this.subscriptions)) { + this.subscriptions.removeIf((sub) -> ObjectUtils.containsElement(topic, sub.getTopic())); + } } catch (MqttException ex) { throw new MessagingException("Failed to unsubscribe from topic(s) " + Arrays.toString(topic), ex); @@ -392,25 +450,20 @@ public class Mqttv5PahoMessageDrivenChannelAdapter this.mqttClient = clientManager.getClient(); } - String[] topics = getTopic(); + MqttSubscription[] mqttSubscriptions = obtainSubscriptions(); + if (ObjectUtils.isEmpty(mqttSubscriptions)) { + return; + } ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher(); this.topicLock.lock(); try { - if (topics.length == 0) { - return; - } - - int[] requestedQos = getQos(); - MqttSubscription[] subscriptions = IntStream.range(0, topics.length) - .mapToObj(i -> new MqttSubscription(topics[i], requestedQos[i])) - .toArray(MqttSubscription[]::new); IMqttMessageListener listener = this::messageArrived; - IMqttMessageListener[] listeners = IntStream.range(0, topics.length) + IMqttMessageListener[] listeners = IntStream.range(0, mqttSubscriptions.length) .mapToObj(t -> listener) .toArray(IMqttMessageListener[]::new); - this.mqttClient.subscribe(subscriptions, null, null, listeners, null) + this.mqttClient.subscribe(mqttSubscriptions, null, null, listeners, null) .waitForCompletion(getCompletionTimeout()); - String message = "Connected and subscribed to " + Arrays.toString(topics); + String message = "Connected and subscribed to " + Arrays.toString(mqttSubscriptions); logger.debug(message); if (applicationEventPublisher != null) { applicationEventPublisher.publishEvent(new MqttSubscribedEvent(this, message)); @@ -420,13 +473,29 @@ public class Mqttv5PahoMessageDrivenChannelAdapter if (applicationEventPublisher != null) { applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, ex)); } - logger.error(ex, () -> "Error subscribing to " + Arrays.toString(topics)); + logger.error(ex, () -> "Error subscribing to " + Arrays.toString(mqttSubscriptions)); } finally { this.topicLock.unlock(); } } + private MqttSubscription[] obtainSubscriptions() { + if (this.subscriptions != null) { + return this.subscriptions.toArray(new MqttSubscription[0]); + } + else { + String[] topics = getTopic(); + if (topics.length == 0) { + return null; + } + int[] requestedQos = getQos(); + return IntStream.range(0, topics.length) + .mapToObj(i -> new MqttSubscription(topics[i], requestedQos[i])) + .toArray(MqttSubscription[]::new); + } + } + @Override public void authPacketArrived(int reasonCode, MqttProperties properties) { diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackTests.java index 427ec5efbe..aa1cba6e39 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,6 +20,7 @@ import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.List; +import org.eclipse.paho.mqttv5.common.MqttSubscription; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -76,13 +77,15 @@ public class Mqttv5BackToBackTests implements MosquittoContainerTest { @Test //GH-3732 public void testNoNpeIsNotThrownInCaseDoInitIsNotInvokedBeforeTopicAddition() { - Mqttv5PahoMessageDrivenChannelAdapter channelAdapter = new Mqttv5PahoMessageDrivenChannelAdapter("tcp://mock-url.com:8091", "mock-client-id", "123"); + Mqttv5PahoMessageDrivenChannelAdapter channelAdapter = + new Mqttv5PahoMessageDrivenChannelAdapter("tcp://mock-url.com:8091", "mock-client-id", "123"); Assertions.assertDoesNotThrow(() -> channelAdapter.addTopic("abc", 1)); } @Test //GH-3732 public void testNoNpeIsNotThrownInCaseDoInitIsNotInvokedBeforeTopicRemoval() { - Mqttv5PahoMessageDrivenChannelAdapter channelAdapter = new Mqttv5PahoMessageDrivenChannelAdapter("tcp://mock-url.com:8091", "mock-client-id", "123"); + Mqttv5PahoMessageDrivenChannelAdapter channelAdapter = + new Mqttv5PahoMessageDrivenChannelAdapter("tcp://mock-url.com:8091", "mock-client-id", "123"); Assertions.assertDoesNotThrow(() -> channelAdapter.removeTopic("abc")); } @@ -171,8 +174,12 @@ public class Mqttv5BackToBackTests implements MosquittoContainerTest { @Bean public IntegrationFlow mqttInFlow() { + MqttSubscription mqttSubscription = new MqttSubscription("siTest", 2); + mqttSubscription.setNoLocal(true); + mqttSubscription.setRetainAsPublished(true); Mqttv5PahoMessageDrivenChannelAdapter messageProducer = - new Mqttv5PahoMessageDrivenChannelAdapter(MosquittoContainerTest.mqttUrl(), "mqttv5SIin", "siTest"); + new Mqttv5PahoMessageDrivenChannelAdapter(MosquittoContainerTest.mqttUrl(), "mqttv5SIin", + mqttSubscription); messageProducer.setPayloadType(String.class); messageProducer.setMessageConverter(mqttStringToBytesConverter()); messageProducer.setManualAcks(true); diff --git a/src/reference/antora/modules/ROOT/pages/mqtt.adoc b/src/reference/antora/modules/ROOT/pages/mqtt.adoc index 4bebcc6444..3c2486d7ee 100644 --- a/src/reference/antora/modules/ROOT/pages/mqtt.adoc +++ b/src/reference/antora/modules/ROOT/pages/mqtt.adoc @@ -465,6 +465,9 @@ The `HeaderMapper` is used to map `PUBLISH` frame properties (in Standard `MqttMessage` properties, such as `qos`, `id`, `dup`, `retained`, plus received topic are always mapped to headers. See `MqttHeaders` for more information. +Starting with version 6.3, the `Mqttv5PahoMessageDrivenChannelAdapter` provides constructors based on the `MqttSubscription` for fine-grained configuration instead of plain topic names. +When these subscriptions are provided, the `qos` option of the channel adapter cannot be used, since such a `qos` mode is a part of `MqttSubscription` API. + The following Java DSL configuration sample demonstrates how to use this channel adapter in the integration flow: [source, java] diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index e6206577f0..aaf9169868 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -23,4 +23,10 @@ In general the project has been moved to the latest dependency versions. === Security Support Changes The `spring-integration-security` module is completely removed (being deprecated previously) in favor of API from `spring-security-messaging` module. -See xref:security.adoc[Security in Spring Integration] for more information. \ No newline at end of file +See xref:security.adoc[Security in Spring Integration] for more information. + +[[x6.3-mqtt]] +=== MQTT Support Changes + +The fine-grained configuration based on `MqttSubscription` API is exposed on the `Mqttv5PahoMessageDrivenChannelAdapter`. +See xref:mqtt.adoc[MQTT Support] for more information. \ No newline at end of file