Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4463: Full access to MqttConnectOptions

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4463

Certain options, such as `maxInFlight` were not exposed.

Deprecate the setters on the factory and allow the user to inject a pre-configured
`MqttConnectOptions`, thus making all (and any new) properties available to be
configured.

# Conflicts:
#	spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
This commit is contained in:
Gary Russell
2018-05-07 11:13:07 -04:00
committed by Artem Bilan
parent 8a7cf6cc65
commit 48e8b14475
4 changed files with 133 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -20,7 +20,11 @@
</int:channel>
<bean id="multiUriClientFactory" class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
<property name="serverURIs" value="tcp://localhost:1883,tcp://localhost:1883"/>
<property name="connectionOptions">
<bean class="org.eclipse.paho.client.mqttv3.MqttConnectOptions">
<property name="serverURIs" value="tcp://localhost:1883,tcp://localhost:1883"/>
</bean>
</property>
</bean>
</beans>

112
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
View File

@@ -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));

16
src/reference/asciidoc/mqtt.adoc
View File

@@ -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:
----
<bean id="clientFactory"
class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
<property name="userName" value="${mqtt.username}"/>
<property name="password" value="${mqtt.password}"/>
<bean class="org.eclipse.paho.client.mqttv3.MqttConnectOptions">
<property name="userName" value="${mqtt.username}"/>
<property name="password" value="${mqtt.password}"/>
</bean>
</bean>
<int-mqtt:message-driven-channel-adapter id="mqttInbound"
@@ -266,9 +270,11 @@ public class MqttJavaApplication {
@Bean
public MqttPahoClientFactory mqttClientFactory() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
factory.setServerURIs("tcp://host1:1883", "tcp://host2:1883");
factory.setUserName("username");
factory.setPassword("password");
MqttConnectOptions options = new MqttConnectOptions();
options.setServerURIs(new String[] { "tcp://host1:1883", "tcp://host2:1883" });
options.setUserName("username");
options.setPassword("password".toCharArray());
factory.setConnectionOptions(options);
return factory;
}