Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-8850: Mqttv5MDChA: MqttSubscription-based config

Browse Source 
Fixes: #8850

Expose `MqttSubscription`-based ctors for the `Mqttv5PahoMessageDrivenChannelAdapter`
This commit is contained in:
Artem Bilan
2024-01-16 17:33:51 -05:00
parent 39c99c0719
commit 710558f68a
4 changed files with 106 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
View File

@@ -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<IMqttAsyncClient, MqttConnectionOptions>
implements MqttCallback, MqttComponent<MqttConnectionOptions> {
private final Lock lock = new ReentrantLock();
private final Lock lock = new ReentrantLock();
private final MqttConnectionOptions connectionOptions;
private List<MqttSubscription> 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<IMqttAsyncClient, MqttConnectionOptions> 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) {

15
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackTests.java
View File

@@ -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);