diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java index 48533624c0..d76b9e1b5b 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttMessageDrivenChannelAdapterParser.java +++ b/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"); diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java index 46cfde44c5..d1b0697f02 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttOutboundChannelAdapterParser.java +++ b/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")) || diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttParserUtils.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttParserUtils.java index d4a24622d5..7489554e35 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/config/xml/MqttParserUtils.java +++ b/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"); diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/DefaultMqttPahoClientFactory.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/DefaultMqttPahoClientFactory.java index 60d72b6619..48286dd900 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/DefaultMqttPahoClientFactory.java +++ b/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; } 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 973b11095c..deb0799457 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 @@ -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"); 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 c4db348019..5549fe2d5e 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 @@ -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()); } 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 601c58b693..d9f0832210 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 @@ -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; diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java index 0460b59f00..6808b7ab76 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java +++ b/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"); diff --git a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-4.1.xsd b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-4.1.xsd index 6025456f77..9e8a5d280c 100644 --- a/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-4.1.xsd +++ b/spring-integration-mqtt/src/main/resources/org/springframework/integration/mqtt/config/xml/spring-integration-mqtt-4.1.xsd @@ -167,7 +167,7 @@ - + MQTT broker URL. diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests-context.xml b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests-context.xml new file mode 100644 index 0000000000..ad8622c558 --- /dev/null +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/BackToBackAdapterTests-context.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + + + 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 73b411721d..7867c8c7a3 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 @@ -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("foo")); + Message message = in.receive(10000); + assertNotNull(message); + assertEquals("foo", message.getPayload()); + } } diff --git a/src/reference/docbook/mqtt.xml b/src/reference/docbook/mqtt.xml index 1b9e1f226e..cdc7d22581 100644 --- a/src/reference/docbook/mqtt.xml +++ b/src/reference/docbook/mqtt.xml @@ -55,7 +55,15 @@ The client id. - The broker URL. + + The broker URL. + + + Starting with version 4.1 the url can be omitted and, instead, + the server URIs can be provided in the serverURIs property of the + DefaultMqttPahoClientFactory. This enables, for example, + connection to a highly available (HA) cluster. + A comma delimited list of topics from which this adapter will receive messages. @@ -112,7 +120,15 @@ The client id. - The broker URL. + + The broker URL. + + + Starting with version 4.1 the url can be omitted and, instead, + the server URIs can be provided in the serverURIs property of the + DefaultMqttPahoClientFactory. This enables, for example, + connection to a highly available (HA) cluster. + An MqttMessageConverter (optional). The default diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 9dace90410..e99b780ef8 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -54,5 +54,13 @@ See for more information. +
+ MQTT Adapter Changes + + The MQTT channel adapters can now be configured to connect to multiple servers, + for example, to support High Availability (HA). + See for more information. + +