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

GH-8720: Check MQTT topics if not empty strings

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/8720

Validate MQTT topics for empty strings in the channel adapters configuration

Use plural names for varargs params

**Cherry-pick to `6.1.x` & `6.0.x`**

# Conflicts:
#	spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java
#	spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java
#	spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
This commit is contained in:
Artem Bilan
2023-09-13 09:43:35 -04:00
parent a2356ec036
commit a82a41ece3
3 changed files with 66 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -78,12 +78,22 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro
Assert.noNullElements(topic, "'topics' cannot have null elements");
this.url = url;
this.clientId = clientId;
validateTopics(topic);
this.topics = new LinkedHashSet<>();
for (String t : topic) {
this.topics.add(new Topic(t, 1));
}
}
private static void validateTopics(String[] topics) {
Assert.notNull(topics, "'topics' cannot be null");
Assert.noNullElements(topics, "'topics' cannot have null elements");
for (String topic : topics) {
Assert.hasText(topic, "The topic to subscribe cannot be empty string");
}
}
public void setConverter(MqttMessageConverter converter) {
Assert.notNull(converter, "'converter' cannot be null");
this.converter = converter;
@@ -207,6 +217,7 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro
*/
@ManagedOperation
public void addTopic(String topic, int qos) {
validateTopics(new String[] {topic});
this.topicLock.lock();
try {
Topic topik = new Topic(topic, qos);
@@ -223,16 +234,16 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro
/**
* Add a topic (or topics) to the subscribed list (qos=1).
* @param topic The topics.
* @throws MessagingException if the topic is already in the list.
* @param topics The topics.
* @throws MessagingException if the topics is already in the list.
* @since 4.1
*/
@ManagedOperation
public void addTopic(String... topic) {
Assert.notNull(topic, "'topic' cannot be null");
public void addTopic(String... topics) {
validateTopics(topics);
this.topicLock.lock();
try {
for (String t : topic) {
for (String t : topics) {
addTopic(t, 1);
}
}
@@ -243,25 +254,24 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro
/**
* Add topics to the subscribed list.
* @param topic The topics.
* @param topics The topics.
* @param qos The qos for each topic.
* @throws MessagingException if a topic is already in the list.
* @throws MessagingException if a topics is already in the list.
* @since 4.1
*/
@ManagedOperation
public void addTopics(String[] topic, int[] qos) {
Assert.notNull(topic, "'topic' cannot be null.");
Assert.noNullElements(topic, "'topic' cannot contain any null elements.");
Assert.isTrue(topic.length == qos.length, "topic and qos arrays must the be the same length.");
public void addTopics(String[] topics, int[] qos) {
validateTopics(topics);
Assert.isTrue(topics.length == qos.length, "topics and qos arrays must the be the same length.");
this.topicLock.lock();
try {
for (String topik : topic) {
if (this.topics.contains(new Topic(topik, 0))) {
throw new MessagingException("Topic '" + topik + "' is already subscribed.");
for (String topic : topics) {
if (this.topics.contains(new Topic(topic, 0))) {
throw new MessagingException("Topic '" + topic + "' is already subscribed.");
}
}
for (int i = 0; i < topic.length; i++) {
addTopic(topic[i], qos[i]);
for (int i = 0; i < topics.length; i++) {
addTopic(topics[i], qos[i]);
}
}
finally {

33
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2021 the original author or authors.
* Copyright 2002-2023 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.mqtt.outbound;
import java.util.concurrent.atomic.AtomicBoolean;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
@@ -107,6 +108,7 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
* @param defaultTopic the default topic.
*/
public void setDefaultTopic(String defaultTopic) {
Assert.hasText(defaultTopic, "'defaultTopic' must not be empty");
this.defaultTopic = defaultTopic;
}
@@ -295,14 +297,17 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
@Override
protected void onInit() {
super.onInit();
if (this.topicProcessor instanceof BeanFactoryAware && getBeanFactory() != null) {
((BeanFactoryAware) this.topicProcessor).setBeanFactory(getBeanFactory());
}
if (this.qosProcessor instanceof BeanFactoryAware && getBeanFactory() != null) {
((BeanFactoryAware) this.qosProcessor).setBeanFactory(getBeanFactory());
}
if (this.retainedProcessor instanceof BeanFactoryAware && getBeanFactory() != null) {
((BeanFactoryAware) this.retainedProcessor).setBeanFactory(getBeanFactory());
BeanFactory beanFactory = getBeanFactory();
if (beanFactory != null) {
if (this.topicProcessor instanceof BeanFactoryAware) {
((BeanFactoryAware) this.topicProcessor).setBeanFactory(beanFactory);
}
if (this.qosProcessor instanceof BeanFactoryAware) {
((BeanFactoryAware) this.qosProcessor).setBeanFactory(beanFactory);
}
if (this.retainedProcessor instanceof BeanFactoryAware) {
((BeanFactoryAware) this.retainedProcessor).setBeanFactory(beanFactory);
}
}
}
@@ -333,11 +338,13 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
protected void handleMessageInternal(Message<?> message) {
Object mqttMessage = this.converter.fromMessage(message, Object.class);
String topic = this.topicProcessor.processMessage(message);
if (topic == null && this.defaultTopic == null) {
throw new IllegalStateException(
"No topic could be determined from the message and no default topic defined");
if (topic == null) {
topic = this.defaultTopic;
}
publish(topic == null ? this.defaultTopic : topic, mqttMessage, message);
Assert.state(topic != null, "No topic could be determined from the message and no default topic defined");
publish(topic, mqttMessage, message);
}
protected abstract void publish(String topic, Object mqttMessage, Message<?> message);

23
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
View File

@@ -62,6 +62,7 @@ import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
import org.springframework.integration.mqtt.event.MqttIntegrationEvent;
import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
@@ -83,6 +84,7 @@ import org.springframework.util.ReflectionUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import static org.assertj.core.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
@@ -468,7 +470,7 @@ public class MqttAdapterTests {
willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any());
IMqttToken token = mock(IMqttToken.class);
given(token.getGrantedQos()).willReturn(new int[]{ 0x80 });
given(token.getGrantedQos()).willReturn(new int[] {0x80});
willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any());
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
@@ -516,7 +518,7 @@ public class MqttAdapterTests {
willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any());
IMqttToken token = mock(IMqttToken.class);
given(token.getGrantedQos()).willReturn(new int[]{ 2, 0 });
given(token.getGrantedQos()).willReturn(new int[] {2, 0});
willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any());
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
@@ -543,6 +545,19 @@ public class MqttAdapterTests {
verify(client).disconnectForcibly(5_000L);
}
@Test
public void emptyTopicNotAllowed() {
assertThatIllegalArgumentException()
.isThrownBy(() ->
new MqttPahoMessageDrivenChannelAdapter("client_id", mock(MqttPahoClientFactory.class), ""))
.withMessage("The topic to subscribe cannot be empty string");
var adapter = new MqttPahoMessageDrivenChannelAdapter("client_id", mock(MqttPahoClientFactory.class), "topic1");
assertThatIllegalArgumentException()
.isThrownBy(() -> adapter.addTopic(""))
.withMessage("The topic to subscribe cannot be empty string");
}
@Test
public void testNoNPEOnReconnectAndStopRaceCondition() throws Exception {
final IMqttClient client = mock(IMqttClient.class);
@@ -597,7 +612,7 @@ public class MqttAdapterTests {
};
MqttConnectOptions connectOptions = new MqttConnectOptions();
connectOptions.setServerURIs(new String[]{ "tcp://localhost:1883" });
connectOptions.setServerURIs(new String[] {"tcp://localhost:1883"});
if (cleanSession != null) {
connectOptions.setCleanSession(cleanSession);
}
@@ -621,7 +636,7 @@ public class MqttAdapterTests {
};
MqttConnectOptions connectOptions = new MqttConnectOptions();
connectOptions.setServerURIs(new String[]{ "tcp://localhost:1883" });
connectOptions.setServerURIs(new String[] {"tcp://localhost:1883"});
factory.setConnectionOptions(connectOptions);
MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("client", factory);
adapter.setDefaultTopic("foo");