From bcc499786cbd6c9941b72f03e2003c582107f131 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 1 Dec 2015 12:10:48 -0500 Subject: [PATCH] INT-3900: MQTT: Don't Unsubscribe if !cleanSession JIRA: https://jira.spring.io/browse/INT-3900 The message-driven adapter should not unsubscribe if `cleanSession` is `false`. (cherry picked from commit 616fc79) --- .../tail/FileTailingMessageProducerTests.java | 2 +- .../mqtt/core/ConsumerStopAction.java | 42 +++++++ .../core/DefaultMqttPahoClientFactory.java | 24 ++++ .../mqtt/core/MqttPahoClientFactory.java | 2 + .../MqttPahoMessageDrivenChannelAdapter.java | 21 +++- .../integration/mqtt/MqttAdapterTests.java | 108 ++++++++++++++++++ src/reference/asciidoc/mqtt.adoc | 20 +++- 7 files changed, 212 insertions(+), 7 deletions(-) create mode 100644 spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/ConsumerStopAction.java diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java index f683b8ffb5..404981e2c2 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java @@ -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()); } diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/ConsumerStopAction.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/ConsumerStopAction.java new file mode 100644 index 0000000000..73f446803d --- /dev/null +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/ConsumerStopAction.java @@ -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 + +} 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 459117ac4d..ff90fd999f 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 @@ -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 diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttPahoClientFactory.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttPahoClientFactory.java index 3308fec038..491b38b0fd 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttPahoClientFactory.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttPahoClientFactory.java @@ -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; /** 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 11b3bf9e9b..d03693decb 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 @@ -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()); 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 7b58baf3f6..28f8908bbb 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 @@ -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(); + } + } diff --git a/src/reference/asciidoc/mqtt.adoc b/src/reference/asciidoc/mqtt.adoc index cc449e6649..993a98b036 100644 --- a/src/reference/asciidoc/mqtt.adoc +++ b/src/reference/asciidoc/mqtt.adoc @@ -35,7 +35,7 @@ A minimal configuration might be: Attributes: -[source] +[source,xml] ---- @@ -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] ----