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 dac3ce4c9e..aacf04f080 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -41,65 +41,97 @@ import org.springframework.util.Assert; */ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory { - private volatile Boolean cleanSession; + private MqttConnectOptions options = new MqttConnectOptions(); - private volatile Integer connectionTimeout; + private MqttClientPersistence persistence; - private volatile Integer keepAliveInterval; - - private volatile String password; - - private volatile SocketFactory socketFactory; - - private volatile Properties sslProperties; - - private volatile String userName; - - private volatile MqttClientPersistence persistence; - - private volatile Will will; - - private volatile String[] serverURIs; - - private volatile ConsumerStopAction consumerStopAction = ConsumerStopAction.UNSUBSCRIBE_CLEAN; + private ConsumerStopAction consumerStopAction = ConsumerStopAction.UNSUBSCRIBE_CLEAN; + /** + * Set the cleanSession. + * @param cleanSession the cleanSession to set. + * @deprecated use {@link #setConnectionOptions(MqttConnectOptions)} instead. + */ + @Deprecated public void setCleanSession(Boolean cleanSession) { - this.cleanSession = cleanSession; + this.options.setCleanSession(cleanSession); } + /** + * Set the connectionTimeout. + * @param connectionTimeout the connectionTimeout to set. + * @deprecated use {@link #setConnectionOptions(MqttConnectOptions)} instead. + */ + @Deprecated public void setConnectionTimeout(Integer connectionTimeout) { - this.connectionTimeout = connectionTimeout; + this.options.setConnectionTimeout(connectionTimeout); } + /** + * Set the keepAliveInterval. + * @param keepAliveInterval the keepAliveInterval to set. + * @deprecated use {@link #setConnectionOptions(MqttConnectOptions)} instead. + */ + @Deprecated public void setKeepAliveInterval(Integer keepAliveInterval) { - this.keepAliveInterval = keepAliveInterval; + this.options.setKeepAliveInterval(keepAliveInterval); } + /** + * Set the password. + * @param password the password to set. + * @deprecated use {@link #setConnectionOptions(MqttConnectOptions)} instead. + */ + @Deprecated public void setPassword(String password) { - this.password = password; + this.options.setPassword(password.toCharArray()); } + /** + * Set the socketFactory. + * @param socketFactory the socketFactory to set. + * @deprecated use {@link #setConnectionOptions(MqttConnectOptions)} instead. + */ + @Deprecated public void setSocketFactory(SocketFactory socketFactory) { - this.socketFactory = socketFactory; + this.options.setSocketFactory(socketFactory); } + /** + * Set the sslProperties. + * @param sslProperties the sslProperties to set. + * @deprecated use {@link #setConnectionOptions(MqttConnectOptions)} instead. + */ + @Deprecated public void setSslProperties(Properties sslProperties) { - this.sslProperties = sslProperties; + this.options.setSSLProperties(sslProperties); } + /** + * Set the userName. + * @param userName the userName to set. + * @deprecated use {@link #setConnectionOptions(MqttConnectOptions)} instead. + */ + @Deprecated public void setUserName(String userName) { - this.userName = userName; + this.options.setUserName(userName); } /** * Will be used to set the "Last Will and Testament" (LWT) for the connection. * @param will The will. * @see MqttConnectOptions#setWill + * @deprecated use {@link #setConnectionOptions(MqttConnectOptions)} instead. */ + @Deprecated public void setWill(Will will) { - this.will = will; + this.options.setWill(will.getTopic(), will.getPayload(), will.getQos(), will.isRetained()); } + /** + * Set the persistence to pass into the client constructor. + * @param persistence the persistence to set. + */ public void setPersistence(MqttClientPersistence persistence) { this.persistence = persistence; } @@ -109,10 +141,12 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory { * @param serverURIs The URIs. * @see MqttConnectOptions#setServerURIs(String[]) * @since 4.1 + * @deprecated use {@link #setConnectionOptions(MqttConnectOptions)} instead. */ + @Deprecated public void setServerURIs(String... serverURIs) { Assert.notNull(serverURIs, "'serverURIs' must not be null."); - this.serverURIs = Arrays.copyOf(serverURIs, serverURIs.length); + this.options.setServerURIs(Arrays.copyOf(serverURIs, serverURIs.length)); } /** @@ -147,37 +181,19 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory { return new MqttAsyncClient(uri == null ? "tcp://NO_URL_PROVIDED" : uri, clientId, this.persistence); } + /** + * Set the preconfigured {@link MqttConnectOptions}. + * @param options the options. + * @since 4.3.16 + */ + public void setConnectionOptions(MqttConnectOptions options) { + Assert.notNull(options, "MqttConnectOptions cannot be null"); + this.options = options; + } + @Override public MqttConnectOptions getConnectionOptions() { - MqttConnectOptions options = new MqttConnectOptions(); - if (this.cleanSession != null) { - options.setCleanSession(this.cleanSession); - } - if (this.connectionTimeout != null) { - options.setConnectionTimeout(this.connectionTimeout); - } - if (this.keepAliveInterval != null) { - options.setKeepAliveInterval(this.keepAliveInterval); - } - if (this.password != null) { - options.setPassword(this.password.toCharArray()); - } - if (this.socketFactory != null) { - options.setSocketFactory(this.socketFactory); - } - if (this.sslProperties != null) { - options.setSSLProperties(this.sslProperties); - } - if (this.userName != null) { - options.setUserName(this.userName); - } - 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; + return this.options; } public static class Will { 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 index 18318a209d..e26f4f2310 100644 --- 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 @@ -20,7 +20,11 @@ - + + + + + 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 3a66b4a089..8f67fe137f 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -80,7 +80,6 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.mqtt.core.ConsumerStopAction; import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory; -import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory.Will; import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent; import org.springframework.integration.mqtt.event.MqttIntegrationEvent; import org.springframework.integration.mqtt.event.MqttSubscribedEvent; @@ -112,51 +111,23 @@ public class MqttAdapterTests { this.alwaysComplete = (IMqttToken) pfb.getObject(); } - @Test - public void testPahoConnectOptions() { - DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); - factory.setCleanSession(false); - factory.setConnectionTimeout(23); - factory.setKeepAliveInterval(45); - factory.setPassword("pass"); - SocketFactory socketFactory = mock(SocketFactory.class); - factory.setSocketFactory(socketFactory); - Properties props = new Properties(); - factory.setSslProperties(props); - factory.setUserName("user"); - Will will = new Will("foo", "bar".getBytes(), 2, true); - factory.setWill(will); - - MqttConnectOptions options = factory.getConnectionOptions(); - - assertEquals(23, options.getConnectionTimeout()); - assertEquals(45, options.getKeepAliveInterval()); - assertEquals("pass", new String(options.getPassword())); - assertSame(socketFactory, options.getSocketFactory()); - assertSame(props, options.getSSLProperties()); - assertEquals("user", options.getUserName()); - assertEquals("foo", options.getWillDestination()); - assertEquals("bar", new String(options.getWillMessage().getPayload())); - assertEquals(2, options.getWillMessage().getQos()); - - } - @Test public void testOutboundOptionsApplied() throws Exception { DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); - factory.setCleanSession(false); - factory.setConnectionTimeout(23); - factory.setKeepAliveInterval(45); - factory.setPassword("pass"); + MqttConnectOptions connectOptions = new MqttConnectOptions(); + connectOptions.setCleanSession(false); + connectOptions.setConnectionTimeout(23); + connectOptions.setKeepAliveInterval(45); + connectOptions.setPassword("pass".toCharArray()); MemoryPersistence persistence = new MemoryPersistence(); factory.setPersistence(persistence); final SocketFactory socketFactory = mock(SocketFactory.class); - factory.setSocketFactory(socketFactory); + connectOptions.setSocketFactory(socketFactory); final Properties props = new Properties(); - factory.setSslProperties(props); - factory.setUserName("user"); - Will will = new Will("foo", "bar".getBytes(), 2, true); - factory.setWill(will); + connectOptions.setSSLProperties(props); + connectOptions.setUserName("user"); + connectOptions.setWill("foo", "bar".getBytes(), 2, true); + factory.setConnectionOptions(connectOptions); factory = spy(factory); final MqttAsyncClient client = mock(MqttAsyncClient.class); @@ -205,19 +176,20 @@ public class MqttAdapterTests { @Test public void testInboundOptionsApplied() throws Exception { DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); - factory.setCleanSession(false); - factory.setConnectionTimeout(23); - factory.setKeepAliveInterval(45); - factory.setPassword("pass"); + MqttConnectOptions connectOptions = new MqttConnectOptions(); + connectOptions.setCleanSession(false); + connectOptions.setConnectionTimeout(23); + connectOptions.setKeepAliveInterval(45); + connectOptions.setPassword("pass".toCharArray()); MemoryPersistence persistence = new MemoryPersistence(); factory.setPersistence(persistence); final SocketFactory socketFactory = mock(SocketFactory.class); - factory.setSocketFactory(socketFactory); + connectOptions.setSocketFactory(socketFactory); final Properties props = new Properties(); - factory.setSslProperties(props); - factory.setUserName("user"); - Will will = new Will("foo", "bar".getBytes(), 2, true); - factory.setWill(will); + connectOptions.setSSLProperties(props); + connectOptions.setUserName("user"); + connectOptions.setWill("foo", "bar".getBytes(), 2, true); + factory.setConnectionOptions(connectOptions); factory = spy(factory); final IMqttClient client = mock(IMqttClient.class); @@ -421,19 +393,19 @@ public class MqttAdapterTests { @Test public void testSubscribeFailure() throws Exception { DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); - factory.setCleanSession(false); - factory.setConnectionTimeout(23); - factory.setKeepAliveInterval(45); - factory.setPassword("pass"); + MqttConnectOptions connectOptions = new MqttConnectOptions(); + connectOptions.setCleanSession(false); + connectOptions.setConnectionTimeout(23); + connectOptions.setKeepAliveInterval(45); + connectOptions.setPassword("pass".toCharArray()); MemoryPersistence persistence = new MemoryPersistence(); factory.setPersistence(persistence); final SocketFactory socketFactory = mock(SocketFactory.class); - factory.setSocketFactory(socketFactory); + connectOptions.setSocketFactory(socketFactory); final Properties props = new Properties(); - factory.setSslProperties(props); - factory.setUserName("user"); - Will will = new Will("foo", "bar".getBytes(), 2, true); - factory.setWill(will); + connectOptions.setSSLProperties(props); + connectOptions.setUserName("user"); + connectOptions.setWill("foo", "bar".getBytes(), 2, true); factory = spy(factory); MqttAsyncClient aClient = mock(MqttAsyncClient.class); @@ -471,19 +443,19 @@ public class MqttAdapterTests { @Test public void testDifferentQos() throws Exception { DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory(); - factory.setCleanSession(false); - factory.setConnectionTimeout(23); - factory.setKeepAliveInterval(45); - factory.setPassword("pass"); + MqttConnectOptions connectOptions = new MqttConnectOptions(); + connectOptions.setCleanSession(false); + connectOptions.setConnectionTimeout(23); + connectOptions.setKeepAliveInterval(45); + connectOptions.setPassword("pass".toCharArray()); MemoryPersistence persistence = new MemoryPersistence(); factory.setPersistence(persistence); final SocketFactory socketFactory = mock(SocketFactory.class); - factory.setSocketFactory(socketFactory); + connectOptions.setSocketFactory(socketFactory); final Properties props = new Properties(); - factory.setSslProperties(props); - factory.setUserName("user"); - Will will = new Will("foo", "bar".getBytes(), 2, true); - factory.setWill(will); + connectOptions.setSSLProperties(props); + connectOptions.setUserName("user"); + connectOptions.setWill("foo", "bar".getBytes(), 2, true); factory = spy(factory); MqttAsyncClient aClient = mock(MqttAsyncClient.class); @@ -526,13 +498,15 @@ public class MqttAdapterTests { } }; - factory.setServerURIs("tcp://localhost:1883"); + MqttConnectOptions connectOptions = new MqttConnectOptions(); + connectOptions.setServerURIs(new String[] { "tcp://localhost:1883" }); if (cleanSession != null) { - factory.setCleanSession(cleanSession); + connectOptions.setCleanSession(cleanSession); } if (action != null) { factory.setConsumerStopAction(action); } + factory.setConnectionOptions(connectOptions); given(client.isConnected()).willReturn(true); MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("client", factory, "foo"); adapter.setApplicationEventPublisher(mock(ApplicationEventPublisher.class)); diff --git a/src/reference/asciidoc/mqtt.adoc b/src/reference/asciidoc/mqtt.adoc index 31c40fffd2..2d2f993c3e 100644 --- a/src/reference/asciidoc/mqtt.adoc +++ b/src/reference/asciidoc/mqtt.adoc @@ -10,6 +10,8 @@ The current implementation uses the http://www.eclipse.org/paho/[Eclipse Paho MQ Configuration of both adapters is achieved using the `DefaultMqttPahoClientFactory`. Refer to the Paho documentation for more information about configuration options. +NOTE: It is preferred to configure an `MqttConnectOptions` object and inject it into the factory, instead of setting the (deprecated) options on the factory itself. + [[mqtt-inbound]] === Inbound (message-driven) Channel Adapter @@ -21,8 +23,10 @@ A minimal configuration might be: ---- - - + + + +