INT-3467 MQTT Modify Subscribed Topics at Runtime

JIRA: https://jira.spring.io/browse/INT-3467

Add methods and managed operations to the inbound
channel adapter to allow the subscribed topics to be
changed at runtime, programmatically, or using a
control-bus.

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<Topic> 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<Topic>();
+		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 + "]";
+		}
+
+	}
+
 }

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) {

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<String>("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");

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests-context.xml

@@ -10,6 +10,13 @@
 
 	<int:channel id="out"/>
 
+	<int-mqtt:message-driven-channel-adapter id="noTopicsAdapter"
+		auto-startup="false"
+		client-id="foo"
+		url="tcp://localhost:1883"
+		client-factory="clientFactory"
+		channel="out" />
+
 	<int-mqtt:message-driven-channel-adapter id="oneTopicAdapter"
 		auto-startup="false"
 		phase="25"

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParserTests.java

@@ -19,6 +19,9 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertSame;
 
+import java.util.Collection;
+import java.util.Iterator;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -40,6 +43,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 @RunWith(SpringJUnit4ClassRunner.class)
 public class MqttMessageDrivenChannelAdapterParserTests {
 
+	@Autowired
+	private MqttPahoMessageDrivenChannelAdapter noTopicsAdapter;
+
 	@Autowired
 	private MqttPahoMessageDrivenChannelAdapter oneTopicAdapter;
 
@@ -61,14 +67,24 @@ public class MqttMessageDrivenChannelAdapterParserTests {
 	@Autowired
 	private MessageChannel errors;
 
+	@Test
+	public void testNoTopics() { // INT-3467 no longer required to have topics
+		assertEquals("tcp://localhost:1883", TestUtils.getPropertyValue(noTopicsAdapter, "url"));
+		assertFalse(TestUtils.getPropertyValue(noTopicsAdapter, "autoStartup", Boolean.class));
+		assertEquals("foo", TestUtils.getPropertyValue(noTopicsAdapter, "clientId"));
+		assertEquals(0, TestUtils.getPropertyValue(noTopicsAdapter, "topics", Collection.class).size());
+		assertSame(out, TestUtils.getPropertyValue(noTopicsAdapter, "outputChannel"));
+		assertSame(clientFactory, TestUtils.getPropertyValue(noTopicsAdapter, "clientFactory"));
+	}
+
 	@Test
 	public void testOneTopic() {
 		assertEquals("tcp://localhost:1883", TestUtils.getPropertyValue(oneTopicAdapter, "url"));
 		assertFalse(TestUtils.getPropertyValue(oneTopicAdapter, "autoStartup", Boolean.class));
 		assertEquals(25, TestUtils.getPropertyValue(oneTopicAdapter, "phase"));
 		assertEquals("foo", TestUtils.getPropertyValue(oneTopicAdapter, "clientId"));
-		assertEquals("bar", TestUtils.getPropertyValue(oneTopicAdapter, "topic", String[].class)[0]);
-		assertEquals(1, TestUtils.getPropertyValue(oneTopicAdapter, "qos", int[].class)[0]);
+		assertEquals("Topic [topic=bar, qos=1]",
+				TestUtils.getPropertyValue(oneTopicAdapter, "topics", Collection.class).iterator().next().toString());
 		assertSame(converter, TestUtils.getPropertyValue(oneTopicAdapter, "converter"));
 		assertEquals(123L, TestUtils.getPropertyValue(oneTopicAdapter, "messagingTemplate.sendTimeout"));
 		assertSame(out, TestUtils.getPropertyValue(oneTopicAdapter, "outputChannel"));
@@ -82,10 +98,9 @@ public class MqttMessageDrivenChannelAdapterParserTests {
 		assertFalse(TestUtils.getPropertyValue(twoTopicsAdapter, "autoStartup", Boolean.class));
 		assertEquals(25, TestUtils.getPropertyValue(twoTopicsAdapter, "phase"));
 		assertEquals("foo", TestUtils.getPropertyValue(twoTopicsAdapter, "clientId"));
-		assertEquals("bar", TestUtils.getPropertyValue(twoTopicsAdapter, "topic", String[].class)[0]);
-		assertEquals("baz", TestUtils.getPropertyValue(twoTopicsAdapter, "topic", String[].class)[1]);
-		assertEquals(0, TestUtils.getPropertyValue(twoTopicsAdapter, "qos", int[].class)[0]);
-		assertEquals(2, TestUtils.getPropertyValue(twoTopicsAdapter, "qos", int[].class)[1]);
+		Iterator<?> 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());
 	}
 
 }

src/reference/docbook/mqtt.xml

@@ -99,6 +99,25 @@
 				containing the failed message and cause.
 			</callout>
 		</calloutlist>
+		<section>
+			<title>Adding/Removing Topics at Runtime</title>
+			<para>
+				Starting with <emphasis>version 4.1</emphasis>, it is possible to programmatically change the topics
+				to which the adapter is subscribed. Methods <code>addTopic()</code> and <code>removeTopic()</code> are
+				provided. When adding topics, you can optionally specify the <code>QoS</code> (default: 1). You can
+				also modify the topics by sending an appropriate message to a <code>&lt;control-bus/&gt;</code> with
+				an appropriate payload: <code>"myMqttAdapter.addTopic('foo', 1)"</code>.
+			</para>
+			<para>
+				Stopping/starting the adapter has no effect on the topic list (it does <emphasis role="bold">not</emphasis>
+				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.
+			</para>
+			<para>
+				Changing the topics while the adapter is stopped (or disconnected from the broker) will take effect
+				the next time a connection is established.
+			</para>
+		</section>
 	</section>
 
 	<section id="mqtt-outbound">

src/reference/docbook/whats-new.xml

@@ -63,11 +63,15 @@
 			</para>
 			<para>
 				The MQTT message-driven channel adapter now supports specifying the QoS setting for each
-				subscription. See <xref linkend="mqtt"/> for more information.
+				subscription. See <xref linkend="mqtt-inbound"/> for more information.
 			</para>
 			<para>
 				The MQTT outbound channel adapter now supports asynchronous sends, avoiding blocking
-				until delivery is confirmed. See <xref linkend="mqtt"/> for more information.
+				until delivery is confirmed. See <xref linkend="mqtt-outbound"/> for more information.
+			</para>
+			<para>
+				It is now possible to programmatically subscribe to and unsubscribe from topics at runtime.
+				See <xref linkend="mqtt-inbound"/> for more information.
 			</para>
 		</section>
 	</section>