INT-3437 MQTT - Support Cluster Connection

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

Note: Tests are still ignored due to hung connection
on close problem.

INT-3437 Polishing - PR Comments

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java

@@ -40,7 +40,7 @@ public class MqttMessageDrivenChannelAdapterParser extends AbstractChannelAdapte
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder
 				.genericBeanDefinition(MqttPahoMessageDrivenChannelAdapter.class);
 
-		MqttParserUtils.parseCommon(element, builder);
+		MqttParserUtils.parseCommon(element, builder, parserContext);
 		builder.addConstructorArgValue(element.getAttribute("topics"));
 		builder.addPropertyReference("outputChannel", channelName);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java

@@ -49,7 +49,7 @@ public class MqttOutboundChannelAdapterParser extends AbstractOutboundChannelAda
 
 		final BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MqttPahoMessageHandler.class);
 
-		MqttParserUtils.parseCommon(element, builder);
+		MqttParserUtils.parseCommon(element, builder, parserContext);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-topic");
 		if (StringUtils.hasText(element.getAttribute("converter")) &&
 				(StringUtils.hasText(element.getAttribute("default-qos")) ||

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttParserUtils.java

@@ -19,6 +19,7 @@ import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.util.StringUtils;
 
@@ -38,13 +39,23 @@ public final class MqttParserUtils {
 		throw new AssertionError();
 	}
 
-	public static void parseCommon(Element element, BeanDefinitionBuilder builder) {
-		builder.addConstructorArgValue(element.getAttribute("url"));
+	public static void parseCommon(Element element, BeanDefinitionBuilder builder, ParserContext parserContext) {
+
+		String url = element.getAttribute("url");
+		if (StringUtils.hasText(url)) {
+			builder.addConstructorArgValue(url);
+		}
 		builder.addConstructorArgValue(element.getAttribute("client-id"));
 		String clientFactory = element.getAttribute("client-factory");
 		if (StringUtils.hasText(clientFactory)) {
 			builder.addConstructorArgReference(clientFactory);
 		}
+		else {
+			if (!StringUtils.hasText(url)) {
+				parserContext.getReaderContext().error("If no 'url' attribute is provided, a 'client-factory' " +
+						"(with serverURIs) is required", element);
+			}
+		}
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "converter");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/DefaultMqttPahoClientFactory.java

@@ -52,6 +52,8 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory {
 
 	private volatile Will will;
 
+	private volatile String[] serverURIs;
+
 	public void setCleanSession(Boolean cleanSession) {
 		this.cleanSession = cleanSession;
 	}
@@ -94,9 +96,20 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory {
 		this.persistence = persistence;
 	}
 
+	/**
+	 * Use this when using multiple server instances, for example when using HA.
+	 * @param serverURIs The URIs.
+	 * @see MqttConnectOptions#setServerURIs(String[])
+	 * @since 4.1
+	 */
+	public void setServerURIs(String[] serverURIs) {
+		this.serverURIs = serverURIs;
+	}
+
 	@Override
-	public MqttClient getClientInstance(String url, String clientId) throws MqttException {
-		return new MqttClient(url, clientId, this.persistence);
+	public MqttClient getClientInstance(String uri, String clientId) throws MqttException {
+		// Client validates URI even if overridden by options
+		return new MqttClient(uri == null ? "tcp://NO_URL_PROVIDED" : uri, clientId, this.persistence);
 	}
 
 	@Override
@@ -126,6 +139,9 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory {
 		if (this.will != null) {
 			options.setWill(this.will.getTopic(), this.will.getPayload(), this.will.getQos(), this.will.isRetained());
 		}
+		if (this.serverURIs != null) {
+			options.setServerURIs(this.serverURIs);
+		}
 		return options;
 	}
 

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

@@ -38,7 +38,6 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro
 	private volatile MqttMessageConverter converter;
 
 	public AbstractMqttMessageDrivenChannelAdapter(String url, String clientId, String... topic) {
-		Assert.hasText(url, "'url' cannot be null or empty");
 		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");

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

@@ -21,12 +21,14 @@ import java.util.concurrent.ScheduledFuture;
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
 import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 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.util.Assert;
 
 /**
  * Eclipse Paho Implementation.
@@ -47,12 +49,41 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 	private volatile boolean connected;
 
 
+	/**
+	 * Use this constructor for a single url (although it may be overridden
+	 * if the server URI(s) are provided by the {@link MqttConnectOptions#getServerURIs()}
+	 * provided by the {@link MqttPahoClientFactory}).
+	 * @param url the URL.
+	 * @param clientId The client id.
+	 * @param clientFactory The client factory.
+	 * @param topic The topic(s).
+	 */
 	public MqttPahoMessageDrivenChannelAdapter(String url, String clientId, MqttPahoClientFactory clientFactory,
 			String... topic) {
 		super(url, clientId, topic);
 		this.clientFactory = clientFactory;
 	}
 
+	/**
+	 * Use this constructor if the server URI(s) are provided by the {@link MqttConnectOptions#getServerURIs()}
+	 * provided by the {@link MqttPahoClientFactory}.
+	 * @param clientId The client id.
+	 * @param clientFactory The client factory.
+	 * @param topic The topic(s).
+	 * @since 4.1
+	 */
+	public MqttPahoMessageDrivenChannelAdapter(String clientId, MqttPahoClientFactory clientFactory,
+			String... topic) {
+		super(null, clientId, topic);
+		this.clientFactory = clientFactory;
+	}
+
+	/**
+	 * Use this URL when you don't need additional {@link MqttConnectOptions}.
+	 * @param url The URL.
+	 * @param clientId The client id.
+	 * @param topic The topic(s).
+	 */
 	public MqttPahoMessageDrivenChannelAdapter(String url, String clientId, String... topic) {
 		this(url, clientId, new DefaultMqttPahoClientFactory(), topic);
 	}
@@ -96,9 +127,13 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 	}
 
 	private void connectAndSubscribe() throws MqttException {
-		this.client = this.clientFactory.getClientInstance(this.getUrl(), this.getClientId());
 		this.client.setCallback(this);
-		this.client.connect(this.clientFactory.getConnectionOptions());
+		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.getClientInstance(this.getUrl(), this.getClientId());
+		this.client.connect(connectionOptions);
+
 		try {
 			this.client.subscribe(this.getTopic());
 		}

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

@@ -55,7 +55,6 @@ public abstract class AbstractMqttMessageHandler extends AbstractMessageHandler
 	private volatile boolean autoStartup;
 
 	public AbstractMqttMessageHandler(String url, String clientId) {
-		Assert.hasText(url, "'url' cannot be null or empty");
 		Assert.hasText(clientId, "'clientId' cannot be null or empty");
 		this.url = url;
 		this.clientId = clientId;

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java

@@ -18,6 +18,7 @@ package org.springframework.integration.mqtt.outbound;
 import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
 import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
 
@@ -40,11 +41,36 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
 
 	private volatile MqttClient client;
 
-	public MqttPahoMessageHandler(String url, String clientId, MqttPahoClientFactory factory) {
+	/**
+	 * Use this constructor for a single url (although it may be overridden
+	 * if the server URI(s) are provided by the {@link MqttConnectOptions#getServerURIs()}
+	 * provided by the {@link MqttPahoClientFactory}).
+	 * @param url the URL.
+	 * @param clientId The client id.
+	 * @param clientFactory The client factory.
+	 */
+	public MqttPahoMessageHandler(String url, String clientId, MqttPahoClientFactory clientFactory) {
 		super(url, clientId);
-		this.clientFactory = factory;
+		this.clientFactory = clientFactory;
 	}
 
+	/**
+	 * Use this constructor if the server URI(s) are provided by the {@link MqttConnectOptions#getServerURIs()}
+	 * provided by the {@link MqttPahoClientFactory}.
+	 * @param clientId The client id.
+	 * @param clientFactory The client factory.
+	 * @since 4.1
+	 */
+	public MqttPahoMessageHandler(String clientId, MqttPahoClientFactory clientFactory) {
+		super(null, clientId);
+		this.clientFactory = clientFactory;
+	}
+
+	/**
+	 * Use this URL when you don't need additional {@link MqttConnectOptions}.
+	 * @param url The URL.
+	 * @param clientId The client id.
+	 */
 	public MqttPahoMessageHandler(String url, String clientId) {
 		this(url, clientId, new DefaultMqttPahoClientFactory());
 	}
@@ -73,8 +99,11 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
 			this.client = null;
 		}
 		if (this.client == null) {
+			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.getClientInstance(this.getUrl(), this.getClientId());
-			this.client.connect(this.clientFactory.getConnectionOptions());
+			this.client.connect(connectionOptions);
 			this.client.setCallback(this);
 			if (logger.isDebugEnabled()) {
 				logger.debug("Client connected");

spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-4.1.xsd

@@ -167,7 +167,7 @@
 				<xsd:union memberTypes="xsd:integer xsd:string" />
 			</xsd:simpleType>
 		</xsd:attribute>
-		<xsd:attribute name="url" use="required" type="xsd:string">
+		<xsd:attribute name="url" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation>
 					MQTT broker URL.

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

@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-mqtt="http://www.springframework.org/schema/integration/mqtt"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/mqtt http://www.springframework.org/schema/integration/mqtt/spring-integration-mqtt.xsd">
+
+	<int-mqtt:outbound-channel-adapter id="out" client-id="multiOut" 
+		client-factory="multiUriClientFactory"
+		default-topic="multiServerTests" />
+
+	<int-mqtt:message-driven-channel-adapter id="inbound"
+		client-factory="multiUriClientFactory"
+		client-id="multiIn" channel="in" topics="multiServerTests" />
+
+	<int:channel id="in">
+		<int:queue />
+	</int:channel>
+
+	<bean id="multiUriClientFactory" class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
+		<property name="serverURIs" value="tcp://localhost:1883,tcp://localhost:1883"/>
+	</bean>
+
+</beans>

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackTobackAdapterTests.java

@@ -22,16 +22,23 @@ import static org.mockito.Mockito.mock;
 import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
+import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
 import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
 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.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Gary Russell
@@ -39,11 +46,20 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
  *
  */
 @Ignore //TODO transiently
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+@DirtiesContext
 public class BackTobackAdapterTests {
 
 	@Rule
 	public final BrokerRunning brokerRunning = BrokerRunning.isRunning(1883);
 
+	@Autowired
+	public MessageChannel out;
+
+	@Autowired
+	public PollableChannel in;
+
 	@Test
 	public void testSingleTopic() {
 		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("tcp://localhost:1883", "si-test-out");
@@ -103,4 +119,11 @@ public class BackTobackAdapterTests {
 		adapter.stop();
 	}
 
+	@Test
+	public void testMultiURIs() {
+		out.send(new GenericMessage<String>("foo"));
+		Message<?> message = in.receive(10000);
+		assertNotNull(message);
+		assertEquals("foo", message.getPayload());
+	}
 }

src/reference/docbook/mqtt.xml

@@ -55,7 +55,15 @@
 				The client id.
 			</callout>
 			<callout arearefs="mqtt-i-02">
-				The broker URL.
+				<para>
+					The broker URL.
+				</para>
+				<note>
+					Starting with <emphasis>version 4.1</emphasis> the url can be omitted and, instead,
+					the server URIs can be provided in the <code>serverURIs</code> property of the
+					<classname>DefaultMqttPahoClientFactory</classname>. This enables, for example,
+					connection to a highly available (HA) cluster.
+				</note>
 			</callout>
 			<callout arearefs="mqtt-i-03">
 				A comma delimited list of topics from which this adapter will receive messages.
@@ -112,7 +120,15 @@
 				The client id.
 			</callout>
 			<callout arearefs="mqtt-o-02">
-				The broker URL.
+				<para>
+					The broker URL.
+				</para>
+				<note>
+					Starting with <emphasis>version 4.1</emphasis> the url can be omitted and, instead,
+					the server URIs can be provided in the <code>serverURIs</code> property of the
+					<classname>DefaultMqttPahoClientFactory</classname>. This enables, for example,
+					connection to a highly available (HA) cluster.
+				</note>
 			</callout>
 			<callout arearefs="mqtt-o-03">
 				An <interfacename>MqttMessageConverter</interfacename> (optional). The default

src/reference/docbook/whats-new.xml

@@ -54,5 +54,13 @@
 				See <xref linkend="http-namespace"/> for more information.
 			</para>
 		</section>
+		<section id="4.1-mqtt-cluster">
+			<title>MQTT Adapter Changes</title>
+			<para>
+				The MQTT channel adapters can now be configured to connect to multiple servers,
+				for example, to support High Availability (HA).
+				See <xref linkend="mqtt"/> for more information.
+			</para>
+		</section>
 	</section>
 </chapter>