From a82a41ece307e1845454fa6d1e6eeebb49aa7c46 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 13 Sep 2023 09:43:35 -0400 Subject: [PATCH] GH-8720: Check MQTT topics if not empty strings 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 --- ...stractMqttMessageDrivenChannelAdapter.java | 44 ++++++++++++------- .../outbound/AbstractMqttMessageHandler.java | 33 ++++++++------ .../integration/mqtt/MqttAdapterTests.java | 23 ++++++++-- 3 files changed, 66 insertions(+), 34 deletions(-) diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java index da2d3f2c7a..79fcb5c9c4 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java @@ -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 { diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java index a87077cd6f..52d0e31338 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java @@ -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); diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java index ad3e88e2cd..637ed10f2e 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java @@ -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");