From b08809190250b5cb8d211434fc3ae23af194fdd9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 6 Feb 2023 11:58:41 -0500 Subject: [PATCH] GH-4002: Deprecate `ConsumerStopAction` in MQTT (#4006) Fixes https://github.com/spring-projects/spring-integration/issues/4002 Turns out the `ConsumerStopAction` was introduced in a point version to mitigate an unsubscription bug and let to preserve a previous behaviour. * Deprecate `ConsumerStopAction` in favor of just `cleanSession` flag in the `MqttConnectOptions` **Cherry-pick to `5.5.x`** --- .../mqtt/core/ConsumerStopAction.java | 9 +++- .../core/DefaultMqttPahoClientFactory.java | 9 +++- .../mqtt/core/MqttPahoClientFactory.java | 5 +- .../MqttPahoMessageDrivenChannelAdapter.java | 16 ++++--- .../integration/mqtt/MqttAdapterTests.java | 47 +++---------------- 5 files changed, 36 insertions(+), 50 deletions(-) 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 index 639b61744c..1bffae3f94 100644 --- 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2023 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. @@ -17,12 +17,17 @@ package org.springframework.integration.mqtt.core; /** - * Action to take regarding subscrptions when consumer stops. + * Action to take regarding subscriptions when consumer stops. * * @author Gary Russell + * * @since 4.2.3 * + * @deprecated since 5.5.17 + * in favor of standard {@link org.eclipse.paho.client.mqttv3.MqttConnectOptions#setCleanSession(boolean)}. + * Will be removed in 6.1.0. */ +@Deprecated public enum ConsumerStopAction { /** 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 c17eed727e..6472173c29 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-2020 the original author or authors. + * Copyright 2002-2023 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,6 +41,7 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory { private MqttClientPersistence persistence; + @SuppressWarnings("deprecation") private ConsumerStopAction consumerStopAction = ConsumerStopAction.UNSUBSCRIBE_CLEAN; /** @@ -55,7 +56,10 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory { * Get the consumer stop action. * @return the consumer stop action. * @since 4.2.3 + * @deprecated since 5.5.17 in favor of standard {@link MqttConnectOptions#setCleanSession(boolean)}. + * Will be removed in 6.1.0. */ + @Deprecated @Override public ConsumerStopAction getConsumerStopAction() { return this.consumerStopAction; @@ -66,7 +70,10 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory { * Default: {@link ConsumerStopAction#UNSUBSCRIBE_CLEAN}. * @param consumerStopAction the consumer stop action. * @since 4.2.3. + * @deprecated since 5.5.17 in favor of standard {@link MqttConnectOptions#setCleanSession(boolean)}. + * Will be removed in 6.1.0. */ + @Deprecated public void setConsumerStopAction(ConsumerStopAction consumerStopAction) { this.consumerStopAction = consumerStopAction; } 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 1a760b39a9..f15e0208ec 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2023 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. @@ -60,7 +60,10 @@ public interface MqttPahoClientFactory { * Get the consumer stop action. * @return the consumer stop action. * @since 4.3 + * @deprecated since 5.5.17 in favor of standard {@link MqttConnectOptions#setCleanSession(boolean)}. + * Will be removed in 6.1.0. */ + @Deprecated ConsumerStopAction getConsumerStopAction(); } 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 5978d74c2c..4143a95d00 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2023 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. @@ -31,7 +31,6 @@ import org.eclipse.paho.client.mqttv3.MqttMessage; import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.acks.SimpleAcknowledgment; -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.core.MqttPahoComponent; @@ -83,7 +82,8 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv private volatile boolean cleanSession; - private volatile ConsumerStopAction consumerStopAction; + @SuppressWarnings("deprecation") + private volatile org.springframework.integration.mqtt.core.ConsumerStopAction consumerStopAction; /** * Use this constructor for a single url (although it may be overridden if the server @@ -184,13 +184,16 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv } } + @SuppressWarnings("deprecation") @Override protected synchronized void doStop() { cancelReconnect(); if (this.client != null) { try { - if (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_ALWAYS) - || (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_CLEAN) + if (this.consumerStopAction + .equals(org.springframework.integration.mqtt.core.ConsumerStopAction.UNSUBSCRIBE_ALWAYS) + || (this.consumerStopAction + .equals(org.springframework.integration.mqtt.core.ConsumerStopAction.UNSUBSCRIBE_CLEAN) && this.cleanSession)) { this.client.unsubscribe(getTopic()); @@ -254,12 +257,13 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv } } + @SuppressWarnings("deprecation") private synchronized void connectAndSubscribe() throws MqttException { // NOSONAR MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions(); this.cleanSession = connectionOptions.isCleanSession(); this.consumerStopAction = this.clientFactory.getConsumerStopAction(); if (this.consumerStopAction == null) { - this.consumerStopAction = ConsumerStopAction.UNSUBSCRIBE_CLEAN; + this.consumerStopAction = org.springframework.integration.mqtt.core.ConsumerStopAction.UNSUBSCRIBE_CLEAN; } Assert.state(getUrl() != null || connectionOptions.getServerURIs() != null, "If no 'url' provided, connectionOptions.getServerURIs() must not be null"); 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 3916a3d7d2..ad3e88e2cd 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 @@ -18,7 +18,6 @@ package org.springframework.integration.mqtt; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; -import java.util.Date; import java.util.Properties; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CountDownLatch; @@ -62,7 +61,6 @@ import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.integration.channel.NullChannel; 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.event.MqttConnectionFailedEvent; import org.springframework.integration.mqtt.event.MqttIntegrationEvent; @@ -124,7 +122,7 @@ public class MqttAdapterTests { public void testCloseOnBadConnectIn() throws Exception { final IMqttClient client = mock(IMqttClient.class); willThrow(new MqttException(0)).given(client).connect(any()); - MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER); + MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null); adapter.start(); verify(client).close(); adapter.stop(); @@ -370,7 +368,7 @@ public class MqttAdapterTests { @Test public void testStopActionDefault() throws Exception { final IMqttClient client = mock(IMqttClient.class); - MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, null); + MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null); adapter.start(); adapter.stop(); @@ -380,35 +378,7 @@ public class MqttAdapterTests { @Test public void testStopActionDefaultNotClean() throws Exception { final IMqttClient client = mock(IMqttClient.class); - MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, false, null); - - adapter.start(); - adapter.stop(); - verifyNotUnsubscribe(client); - } - - @Test - public void testStopActionAlways() throws Exception { - final IMqttClient client = mock(IMqttClient.class); - MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, false, - ConsumerStopAction.UNSUBSCRIBE_ALWAYS); - - adapter.start(); - adapter.stop(); - verifyUnsubscribe(client); - - adapter.connectionLost(new RuntimeException("Intentional")); - - TaskScheduler taskScheduler = TestUtils.getPropertyValue(adapter, "taskScheduler", TaskScheduler.class); - - verify(taskScheduler, never()) - .schedule(any(Runnable.class), any(Date.class)); - } - - @Test - public void testStopActionNever() throws Exception { - final IMqttClient client = mock(IMqttClient.class); - MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER); + MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, false); adapter.start(); adapter.stop(); @@ -439,7 +409,7 @@ public class MqttAdapterTests { @Test public void testReconnect() throws Exception { final IMqttClient client = mock(IMqttClient.class); - MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER); + MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null); adapter.setRecoveryInterval(10); LogAccessor logger = spy(TestUtils.getPropertyValue(adapter, "logger", LogAccessor.class)); new DirectFieldAccessor(adapter).setPropertyValue("logger", logger); @@ -576,7 +546,7 @@ public class MqttAdapterTests { @Test public void testNoNPEOnReconnectAndStopRaceCondition() throws Exception { final IMqttClient client = mock(IMqttClient.class); - MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER); + MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null); adapter.setRecoveryInterval(10); MqttException mqttException = new MqttException(MqttException.REASON_CODE_SUBSCRIBE_FAILED); @@ -615,8 +585,8 @@ public class MqttAdapterTests { taskScheduler.destroy(); } - private MqttPahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttClient client, Boolean cleanSession, - ConsumerStopAction action) { + private MqttPahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttClient client, Boolean cleanSession) + throws MqttException { DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() { @@ -631,9 +601,6 @@ public class MqttAdapterTests { if (cleanSession != null) { connectOptions.setCleanSession(cleanSession); } - if (action != null) { - factory.setConsumerStopAction(action); - } factory.setConnectionOptions(connectOptions); given(client.isConnected()).willReturn(true); MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("client", factory, "foo");