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

INT-3900: MQTT: Don't Unsubscribe if !cleanSession

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

The message-driven adapter should not unsubscribe if `cleanSession` is
`false`.
(cherry picked from commit 616fc79)
This commit is contained in:
Gary Russell
2015-12-01 12:10:48 -05:00
committed by Artem Bilan
parent 373bdfc617
commit bcc499786c
7 changed files with 212 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java
View File

@@ -161,7 +161,7 @@ public class FileTailingMessageProducerTests {
foo.flush();
foo.close();
for (int i = 0; i < 50; i++) {
Message<?> message = outputChannel.receive(5000);
Message<?> message = outputChannel.receive(10000);
assertNotNull("expected a non-null message", message);
assertEquals("hello" + i, message.getPayload());
}

42
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/ConsumerStopAction.java Normal file
View File

@@ -0,0 +1,42 @@
/*
* Copyright 2015 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.mqtt.core;
/**
* Action to take regarding subscrptions when consumer stops.
*
* @author Gary Russell
* @since 4.2.3
*
*/
public enum ConsumerStopAction {
/**
* Never unsubscribe.
*/
UNSUBSCRIBE_NEVER,
/**
* Always unsubscribe.
*/
UNSUBSCRIBE_ALWAYS,
/**
* Unsubscribe if clean session is true.
*/
UNSUBSCRIBE_CLEAN
}

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

@@ -58,6 +58,8 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory {
private volatile String[] serverURIs;
private volatile ConsumerStopAction consumerStopAction = ConsumerStopAction.UNSUBSCRIBE_CLEAN;
public void setCleanSession(Boolean cleanSession) {
this.cleanSession = cleanSession;
}
@@ -110,6 +112,28 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory {
this.serverURIs = Arrays.copyOf(serverURIs, serverURIs.length);
}
/*
* TODO: move to interface in 4.3.
*/
/**
* Get the consumer stop action.
* @return the consumer stop action.
* @since 4.2.3
*/
public ConsumerStopAction getConsumerStopAction() {
return consumerStopAction;
}
/**
* Set the consumer stop action. Determines whether we unsubscribe when the consumer stops.
* Default: {@link ConsumerStopAction#UNSUBSCRIBE_CLEAN}.
* @param consumerStopAction the consumer stop action.
* @since 4.2.3.
*/
public void setConsumerStopAction(ConsumerStopAction consumerStopAction) {
this.consumerStopAction = consumerStopAction;
}
@Override
public MqttClient getClientInstance(String uri, String clientId) throws MqttException {
// Client validates URI even if overridden by options

2
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttPahoClientFactory.java
View File

@@ -35,6 +35,7 @@ public interface MqttPahoClientFactory {
* @return The client instance.
* @throws MqttException Any.
*/
// TODO: change return type to IMqttClient in 4.3
MqttClient getClientInstance(String url, String clientId) throws MqttException;
/**
@@ -46,6 +47,7 @@ public interface MqttPahoClientFactory {
* @throws MqttException Any.
* @since 4.1
*/
// TODO: change return type to IMqttAsyncClient in 4.3
MqttAsyncClient getAsyncClientInstance(String url, String clientId) throws MqttException;
/**

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

@@ -18,8 +18,8 @@ package org.springframework.integration.mqtt.inbound;
import java.util.Arrays;
import java.util.concurrent.ScheduledFuture;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
@@ -27,6 +27,7 @@ import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.integration.mqtt.core.ConsumerStopAction;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
@@ -51,7 +52,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
private final MqttPahoClientFactory clientFactory;
private volatile MqttAsyncClient client;
private volatile IMqttAsyncClient client;
private volatile ScheduledFuture<?> reconnectFuture;
@@ -61,6 +62,10 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
private volatile int recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
private volatile boolean cleanSession;
private volatile ConsumerStopAction consumerStopAction;
private ApplicationEventPublisher applicationEventPublisher;
/**
@@ -148,8 +153,12 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
super.doStop();
if (this.client != null) {
try {
this.client.unsubscribe(getTopic())
.waitForCompletion(this.completionTimeout);
if (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_ALWAYS)
|| (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_CLEAN)
&& this.cleanSession)) {
this.client.unsubscribe(getTopic())
.waitForCompletion(this.completionTimeout);
}
}
catch (MqttException e) {
logger.error("Exception while unsubscribing", e);
@@ -211,6 +220,10 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
private void connectAndSubscribe() throws MqttException {
MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions();
this.cleanSession = connectionOptions.isCleanSession();
this.consumerStopAction = this.clientFactory instanceof DefaultMqttPahoClientFactory ?
((DefaultMqttPahoClientFactory) this.clientFactory).getConsumerStopAction() :
ConsumerStopAction.UNSUBSCRIBE_CLEAN;
Assert.state(getUrl() != null || connectionOptions.getServerURIs() != null,
"If no 'url' provided, connectionOptions.getServerURIs() must not be null");
this.client = this.clientFactory.getAsyncClientInstance(getUrl(), getClientId());

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

@@ -26,6 +26,7 @@ import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -41,21 +42,28 @@ import java.util.concurrent.atomic.AtomicReference;
import javax.net.SocketFactory;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.MqttToken;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.aop.framework.ProxyFactoryBean;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.QueueChannel;
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;
@@ -74,6 +82,22 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
*/
public class MqttAdapterTests {
private IMqttToken alwaysComplete;
{
ProxyFactoryBean pfb = new ProxyFactoryBean();
pfb.addAdvice(new MethodInterceptor() {
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
return null;
}
});
pfb.setInterfaces(IMqttToken.class);
this.alwaysComplete = (IMqttToken) pfb.getObject();
}
@Test
public void testPahoConnectOptions() {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
@@ -313,4 +337,88 @@ public class MqttAdapterTests {
assertEquals("Connected and subscribed to [baz, fix]", ((MqttSubscribedEvent) event).getMessage());
}
@Test
public void testStopActionDefault() throws Exception {
final MqttAsyncClient client = mock(MqttAsyncClient.class);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, null);
adapter.start();
adapter.stop();
verifyUnsubscribe(client);
}
@Test
public void testStopActionDefaultNotClean() throws Exception {
final MqttAsyncClient client = mock(MqttAsyncClient.class);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, false, null);
adapter.start();
adapter.stop();
verifyNotUnsubscribe(client);
}
@Test
public void testStopActionAlways() throws Exception {
final MqttAsyncClient client = mock(MqttAsyncClient.class);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, false,
ConsumerStopAction.UNSUBSCRIBE_ALWAYS);
adapter.start();
adapter.stop();
verifyUnsubscribe(client);
}
@Test
public void testStopActionNever() throws Exception {
final MqttAsyncClient client = mock(MqttAsyncClient.class);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
adapter.start();
adapter.stop();
verifyNotUnsubscribe(client);
}
private MqttPahoMessageDrivenChannelAdapter buildAdapter(final MqttAsyncClient client, Boolean cleanSession,
ConsumerStopAction action) throws MqttException, MqttSecurityException {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() {
@Override
public MqttAsyncClient getAsyncClientInstance(String uri, String clientId) throws MqttException {
return client;
}
};
factory.setServerURIs("tcp://localhost:1883");
if (cleanSession != null) {
factory.setCleanSession(cleanSession);
}
if (action != null) {
factory.setConsumerStopAction(action);
}
when(client.connect(any(MqttConnectOptions.class))).thenReturn(this.alwaysComplete);
when(client.subscribe(any(String[].class), any(int[].class))).thenReturn(this.alwaysComplete);
when(client.disconnect()).thenReturn(this.alwaysComplete);
when(client.unsubscribe(any(String[].class))).thenReturn(this.alwaysComplete);
when(client.isConnected()).thenReturn(true);
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("client", factory, "foo");
adapter.setApplicationEventPublisher(mock(ApplicationEventPublisher.class));
adapter.setOutputChannel(new NullChannel());
adapter.afterPropertiesSet();
return adapter;
}
private void verifyUnsubscribe(MqttAsyncClient client) throws Exception {
verify(client).connect(any(MqttConnectOptions.class));
verify(client).subscribe(any(String[].class), any(int[].class));
verify(client).unsubscribe(any(String[].class));
verify(client).disconnect();
}
private void verifyNotUnsubscribe(MqttAsyncClient client) throws Exception {
verify(client).connect(any(MqttConnectOptions.class));
verify(client).subscribe(any(String[].class), any(int[].class));
verify(client, never()).unsubscribe(any(String[].class));
verify(client).disconnect();
}
}

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

@@ -35,7 +35,7 @@ A minimal configuration might be:
Attributes:
[source]
[source,xml]
----
<int-mqtt:message-driven-channel-adapter id="oneTopicAdapter"
client-id="foo" <1>
@@ -89,6 +89,22 @@ Also, a new property `recoveryInterval` controls the interval at which the adapt
a failure; it defaults to `10000ms` (ten seconds).
This is not currently available using XML configuration.
[NOTE]
====
Prior to _version 4.2.3_, the client always unsubscribed when the adapter was stopped.
This is incorrect because if the client QOS is > 1, we need to keep the subscription active so that messages arriving
while the adapter is stopped will be delivered on the next start.
This also requires setting the `cleanSession` property on the client factory to `false` - it defaults to `true`.
Starting with _version 4.2.3_, the adapter will not unsubscribe (by default) if the `cleanSession` property is `false`.
This behavior can be overridden by setting the `consumerCloseAction` property on the factory.
It can have values: `UNSUBSCRIBE_ALWAYS`, `UNSUBSCRIBE_NEVER`, and `UNSUBSCRIBE_CLEAN`.
The latter (the default) will unsubscribe only if the `cleanSession` property is `true`.
To revert to the pre-4.2.3 behavior, use `UNSUBSCRIBE_ALWAYS`.
====
==== Adding/Removing Topics at Runtime
Starting with _version 4.1_, it is possible to programmatically change the topics to which the adapter is subscribed.
@@ -159,7 +175,7 @@ Starting with _version 4.1_, the adapter supports asynchronous sends, avoiding b
Attributes:
[source]
[source,xml]
----
<int-mqtt:outbound-channel-adapter id="withConverter"
client-id="foo" <1>