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 3b1ffdaded..0dd3bf4210 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 @@ -15,9 +15,18 @@ */ package org.springframework.integration.mqtt.inbound; +import java.util.LinkedHashSet; +import java.util.Set; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; + import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; import org.springframework.integration.mqtt.support.MqttMessageConverter; +import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.jmx.export.annotation.ManagedOperation; +import org.springframework.jmx.export.annotation.ManagedResource; +import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; /** @@ -27,36 +36,29 @@ import org.springframework.util.Assert; * @since 4.0 * */ +@ManagedResource public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessageProducerSupport { private final String url; private final String clientId; - private final String[] topic; - - private volatile int[] qos; + private final Set topics; private volatile MqttMessageConverter converter; + protected final Lock topicLock = new ReentrantLock(); + public AbstractMqttMessageDrivenChannelAdapter(String url, String clientId, String... topic) { Assert.hasText(clientId, "'clientId' cannot be null or empty"); Assert.notNull(topic, "'topics' cannot be null"); - Assert.isTrue(topic.length > 0, "'topics' cannot be empty"); Assert.noNullElements(topic, "'topics' cannot have null elements"); this.url = url; this.clientId = clientId; - this.topic = topic; - // set the topic qos to 1 by default - this.qos = buildQosArray(1); - } - - private int[] buildQosArray(int value) { - int[] qos = new int[this.topic.length]; - for (int i = 0; i < qos.length; i++) { - qos[i] = value; + this.topics = new LinkedHashSet(); + for (String t : topic) { + this.topics.add(new Topic(t, 1)); } - return qos; } public void setConverter(MqttMessageConverter converter) { @@ -73,16 +75,34 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro public void setQos(int... qos) { Assert.notNull(qos, "'qos' cannot be null"); if (qos.length == 1) { - this.qos = buildQosArray(qos[0]); + for (Topic topic : this.topics) { + topic.setQos(qos[0]); + } } else { - Assert.isTrue(qos.length == this.topic.length); - this.qos = qos; + Assert.isTrue(qos.length == this.topics.size(), + "When setting qos, the array must be the same length as the topics"); + int n = 0; + for (Topic topic : this.topics) { + topic.setQos(qos[n++]); + } } } - protected int[] getQos() { - return qos; + @ManagedAttribute + public int[] getQos() { + this.topicLock.lock(); + try { + int[] topicQos = new int[this.topics.size()]; + int n = 0; + for (Topic topic : this.topics) { + topicQos[n++] = topic.getQos(); + } + return topicQos; + } + finally { + this.topicLock.unlock(); + } } protected String getUrl() { @@ -97,8 +117,20 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro return converter; } - protected String[] getTopic() { - return topic; + @ManagedAttribute + public String[] getTopic() { + this.topicLock.lock(); + try { + String[] topicNames = new String[this.topics.size()]; + int n = 0; + for (Topic topic : this.topics) { + topicNames[n++] = topic.getTopic(); + } + return topicNames; + } + finally { + this.topicLock.unlock(); + } } @Override @@ -106,6 +138,102 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro return "mqtt:inbound-channel-adapter"; } + /** + * Add a topic to the subscribed list. + * @param topic The topic. + * @param qos The qos. + * @throws MessagingException if the topic is already in the list. + * @since 4.1 + */ + @ManagedOperation + public void addTopic(String topic, int qos) { + this.topicLock.lock(); + try { + Topic topik = new Topic(topic, qos); + if (this.topics.contains(topik)) { + throw new MessagingException("Topic '" + topic + "' is already subscribed."); + } + this.topics.add(topik); + if (this.logger.isDebugEnabled()) { + logger.debug("Added '" + topic + "' to subscriptions."); + } + } + finally { + this.topicLock.unlock(); + } + } + + /** + * 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. + * @since 4.1 + */ + @ManagedOperation + public void addTopic(String... topic) { + Assert.notNull(topic, "'topic' cannot be null"); + this.topicLock.lock(); + try { + for (String t : topic) { + addTopic(t, 1); + } + } + finally { + this.topicLock.unlock(); + } + } + + /** + * Add topics to the subscribed list. + * @param topic The topics. + * @param qos The qos for each topic. + * @throws MessagingException if a topic 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."); + 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 (int i = 0; i < topic.length; i++) { + addTopic(topic[i], qos[i]); + } + } + finally { + this.topicLock.unlock(); + } + } + + /** + * Remove a topic (or topics) from the subscribed list. + * @param topic The topic. + * @throws MessagingException if the topic is not in the list. + * @since 4.1 + */ + @ManagedOperation + public void removeTopic(String... topic) { + this.topicLock.lock(); + try { + for (String t : topic) { + if (this.topics.remove(new Topic(t, 0))) { + if (this.logger.isDebugEnabled()) { + logger.debug("Removed '" + t + "' from subscriptions."); + } + } + } + } + finally { + this.topicLock.unlock(); + } + } + @Override protected void onInit() { super.onInit(); @@ -114,4 +242,66 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro } } + + /** + * @since 4.1 + */ + private static class Topic { + + private final String topic; + + private volatile int qos; + + public Topic(String topic, int qos) { + this.topic = topic; + this.qos = qos; + } + + public int getQos() { + return qos; + } + + public void setQos(int qos) { + this.qos = qos; + } + + public String getTopic() { + return topic; + } + + @Override + public int hashCode() { + return topic.hashCode(); + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + Topic other = (Topic) obj; + if (topic == null) { + if (other.topic != null) { + return false; + } + } + else if (!topic.equals(other.topic)) { + return false; + } + return true; + } + + @Override + public String toString() { + return "Topic [topic=" + topic + ", qos=" + qos + "]"; + } + + } + } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java index 9fef19f194..9c1db20a6c 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java @@ -28,6 +28,7 @@ import org.eclipse.paho.client.mqttv3.MqttMessage; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; import org.springframework.integration.mqtt.core.MqttPahoClientFactory; import org.springframework.messaging.Message; +import org.springframework.messaging.MessagingException; import org.springframework.util.Assert; /** @@ -142,24 +143,66 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv this.client = null; } + @Override + public void addTopic(String topic, int qos) { + this.topicLock.lock(); + try { + super.addTopic(topic, qos); + if (this.client != null && this.client.isConnected()) { + this.client.subscribe(topic, qos) + .waitForCompletion(this.completionTimeout); + } + } + catch (MqttException e) { + super.removeTopic(topic); + throw new MessagingException("Failed to subscribe to topic " + topic, e); + } + finally { + this.topicLock.unlock(); + } + } + + @Override + public void removeTopic(String... topic) { + this.topicLock.lock(); + try { + if (this.client != null && this.client.isConnected()) { + this.client.unsubscribe(topic) + .waitForCompletion(this.completionTimeout); + } + super.removeTopic(topic); + } + catch (MqttException e) { + throw new MessagingException("Failed to unsubscribe from topic " + Arrays.asList(topic), e); + } + finally { + this.topicLock.unlock(); + } + } + private void connectAndSubscribe() throws MqttException { MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions(); Assert.state(this.getUrl() != null || connectionOptions.getServerURIs() != null, "If no 'url' provided, connectionOptions.getServerURIs() must not be null"); this.client = this.clientFactory.getAsyncClientInstance(this.getUrl(), this.getClientId()); this.client.setCallback(this); - this.client.connect(connectionOptions) - .waitForCompletion(this.completionTimeout); + + this.topicLock.lock(); try { + this.client.connect(connectionOptions) + .waitForCompletion(this.completionTimeout); this.client.subscribe(this.getTopic(), this.getQos()) .waitForCompletion(this.completionTimeout); } catch (MqttException e) { - logger.error("Error subscribing to " + Arrays.asList(this.getTopic()), e); + logger.error("Error connecting or subscribing to " + Arrays.asList(this.getTopic()), e); this.client.disconnect() .waitForCompletion(this.completionTimeout); throw e; } + finally { + this.topicLock.unlock(); + } if (this.client.isConnected()) { this.connected = true; if (this.reconnectFuture != null) { diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java index 62bfa212b1..737b5914dc 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests.java @@ -18,6 +18,7 @@ package org.springframework.integration.mqtt; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -47,6 +48,7 @@ import org.springframework.integration.mqtt.support.MqttHeaders; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -99,6 +101,58 @@ public class BackToBackAdapterTests { assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); } + @Test + public void testAddRemoveTopic() { + MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); + adapter.setDefaultTopic("mqtt-foo"); + adapter.setBeanFactory(mock(BeanFactory.class)); + adapter.afterPropertiesSet(); + adapter.start(); + MqttPahoMessageDrivenChannelAdapter inbound = new MqttPahoMessageDrivenChannelAdapter("tcp://localhost:1883", "si-test-in"); + QueueChannel outputChannel = new QueueChannel(); + inbound.setOutputChannel(outputChannel); + ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); + taskScheduler.initialize(); + inbound.setTaskScheduler(taskScheduler); + inbound.setBeanFactory(mock(BeanFactory.class)); + inbound.afterPropertiesSet(); + inbound.start(); + inbound.addTopic("mqtt-foo"); + adapter.handleMessage(new GenericMessage("foo")); + Message out = outputChannel.receive(10000); + assertNotNull(out); + assertEquals("foo", out.getPayload()); + assertEquals("mqtt-foo", out.getHeaders().get(MqttHeaders.TOPIC)); + + inbound.addTopic("mqtt-bar"); + adapter.handleMessage(MessageBuilder.withPayload("bar").setHeader(MqttHeaders.TOPIC, "mqtt-bar").build()); + out = outputChannel.receive(10000); + assertNotNull(out); + assertEquals("bar", out.getPayload()); + assertEquals("mqtt-bar", out.getHeaders().get(MqttHeaders.TOPIC)); + + inbound.removeTopic("mqtt-bar"); + adapter.handleMessage(MessageBuilder.withPayload("bar").setHeader(MqttHeaders.TOPIC, "mqtt-bar").build()); + out = outputChannel.receive(1000); + assertNull(out); + + try { + inbound.addTopic("mqtt-foo"); + fail("Expected exception"); + } + catch (MessagingException e) { + assertEquals("Topic 'mqtt-foo' is already subscribed.", e.getMessage()); + } + + inbound.addTopic("mqqt-bar", "mqqt-baz"); + inbound.removeTopic("mqqt-bar", "mqqt-baz"); + inbound.addTopics(new String[] { "mqqt-bar", "mqqt-baz" }, new int[] { 0, 0 }); + inbound.removeTopic("mqqt-bar", "mqqt-baz"); + + adapter.stop(); + inbound.stop(); + } + @Test public void testTwoTopics() { MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out"); diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests-context.xml b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests-context.xml index e30bb881bf..d1f5ce060d 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests-context.xml +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests-context.xml @@ -10,6 +10,13 @@ + + iterator = TestUtils.getPropertyValue(twoTopicsAdapter, "topics", Collection.class).iterator(); + assertEquals("Topic [topic=bar, qos=0]", iterator.next().toString()); + assertEquals("Topic [topic=baz, qos=2]", iterator.next().toString()); assertSame(converter, TestUtils.getPropertyValue(twoTopicsAdapter, "converter")); assertEquals(123L, TestUtils.getPropertyValue(twoTopicsAdapter, "messagingTemplate.sendTimeout")); assertSame(out, TestUtils.getPropertyValue(twoTopicsAdapter, "outputChannel")); @@ -94,10 +109,9 @@ public class MqttMessageDrivenChannelAdapterParserTests { @Test public void testTwoTopicsSingleQos() { - assertEquals("bar", TestUtils.getPropertyValue(twoTopicsSingleQosAdapter, "topic", String[].class)[0]); - assertEquals("baz", TestUtils.getPropertyValue(twoTopicsSingleQosAdapter, "topic", String[].class)[1]); - assertEquals(0, TestUtils.getPropertyValue(twoTopicsSingleQosAdapter, "qos", int[].class)[0]); - assertEquals(0, TestUtils.getPropertyValue(twoTopicsSingleQosAdapter, "qos", int[].class)[1]); + Iterator iterator = TestUtils.getPropertyValue(twoTopicsSingleQosAdapter, "topics", Collection.class).iterator(); + assertEquals("Topic [topic=bar, qos=0]", iterator.next().toString()); + assertEquals("Topic [topic=baz, qos=0]", iterator.next().toString()); } } diff --git a/src/reference/docbook/mqtt.xml b/src/reference/docbook/mqtt.xml index 764e8a0162..68d66716c6 100644 --- a/src/reference/docbook/mqtt.xml +++ b/src/reference/docbook/mqtt.xml @@ -99,6 +99,25 @@ containing the failed message and cause. +
+ Adding/Removing Topics at Runtime + + Starting with version 4.1, it is possible to programmatically change the topics + to which the adapter is subscribed. Methods addTopic() and removeTopic() are + provided. When adding topics, you can optionally specify the QoS (default: 1). You can + also modify the topics by sending an appropriate message to a <control-bus/> with + an appropriate payload: "myMqttAdapter.addTopic('foo', 1)". + + + Stopping/starting the adapter has no effect on the topic list (it does not + revert to the original settings in the configuration). The changes are not retained beyond the + life cycle of the application context; a new application context will revert to the configured settings. + + + Changing the topics while the adapter is stopped (or disconnected from the broker) will take effect + the next time a connection is established. + +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 5f37c4bbc2..b96ba11e6f 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -63,11 +63,15 @@ The MQTT message-driven channel adapter now supports specifying the QoS setting for each - subscription. See for more information. + subscription. See for more information. The MQTT outbound channel adapter now supports asynchronous sends, avoiding blocking - until delivery is confirmed. See for more information. + until delivery is confirmed. See for more information. + + + It is now possible to programmatically subscribe to and unsubscribe from topics at runtime. + See for more information.