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

GH-4007: Remove MQTT ConsumerStopAction

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/4007

It was deprecated in the previous version and fully covered with existing `cleanSession` connection option.
This commit is contained in:
abilan
2023-02-17 15:57:04 -05:00
parent 53d1ecd208
commit 7d0265845a
4 changed files with 1 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/ConsumerStopAction.java
View File

@@ -1,48 +0,0 @@
/*
* 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.
* You may obtain a copy of the License at
*
* https://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 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 {
/**
* Never unsubscribe.
*/
UNSUBSCRIBE_NEVER,
/**
* Always unsubscribe.
*/
UNSUBSCRIBE_ALWAYS,
/**
* Unsubscribe if clean session is true.
*/
UNSUBSCRIBE_CLEAN
}

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

@@ -41,9 +41,6 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory {
private MqttClientPersistence persistence;
@SuppressWarnings("deprecation")
private ConsumerStopAction consumerStopAction = ConsumerStopAction.UNSUBSCRIBE_CLEAN;
/**
* Set the persistence to pass into the client constructor.
* @param persistence the persistence to set.
@@ -52,32 +49,6 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory {
this.persistence = persistence;
}
/**
* 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;
}
/**
* 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.
* @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;
}
@Override
public IMqttClient getClientInstance(String uri, String clientId) throws MqttException {
// Client validates URI even if overridden by options

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

@@ -56,14 +56,4 @@ public interface MqttPahoClientFactory {
*/
MqttConnectOptions getConnectionOptions();
/**
* 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();
}

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

@@ -68,9 +68,6 @@ public class MqttPahoMessageDrivenChannelAdapter
private volatile IMqttAsyncClient client;
@SuppressWarnings("deprecation")
private volatile org.springframework.integration.mqtt.core.ConsumerStopAction consumerStopAction;
private volatile boolean readyToSubscribeOnStart;
/**
@@ -184,11 +181,6 @@ public class MqttPahoMessageDrivenChannelAdapter
@SuppressWarnings("deprecation")
private synchronized void connect() throws MqttException {
MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions();
this.consumerStopAction = this.clientFactory.getConsumerStopAction();
if (this.consumerStopAction == null) {
this.consumerStopAction = org.springframework.integration.mqtt.core.ConsumerStopAction.UNSUBSCRIBE_CLEAN;
}
var clientManager = getClientManager();
if (clientManager == null) {
Assert.state(getUrl() != null || connectionOptions.getServerURIs() != null,
@@ -203,17 +195,11 @@ public class MqttPahoMessageDrivenChannelAdapter
}
}
@SuppressWarnings("deprecation")
@Override
protected synchronized void doStop() {
this.readyToSubscribeOnStart = false;
try {
if (this.consumerStopAction
.equals(org.springframework.integration.mqtt.core.ConsumerStopAction.UNSUBSCRIBE_ALWAYS)
|| (this.consumerStopAction
.equals(org.springframework.integration.mqtt.core.ConsumerStopAction.UNSUBSCRIBE_CLEAN)
&& this.clientFactory.getConnectionOptions().isCleanSession())) {
if (this.clientFactory.getConnectionOptions().isCleanSession()) {
this.client.unsubscribe(getTopic());
// Have to re-subscribe on next start if connection is not lost.
this.readyToSubscribeOnStart = true;