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

GH-9112: Workaround for Paho stopReconnectCycle

Browse Source 
Fixes: #9112

`Mqttv5ClientManager` hangs in `stop()` if never was connected.
The scheduled reconnect timer in the client is never cancelled.

* Call `stopReconnectCycle()` on the client via reflection when we disconnect
from the client in Spring Integration MQTT components

**Auto-cherry-pick to `6.2.x` & `6.1.x`**
This commit is contained in:
Artem Bilan
2024-05-21 13:14:56 -04:00
parent da29e2da6a
commit 937da13bcf
11 changed files with 151 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/aot/MqttRuntimeHints.java Normal file
View File

@@ -0,0 +1,57 @@
/*
* Copyright 2024 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.aot;
import java.util.stream.Stream;
import org.springframework.aot.hint.ExecutableMode;
import org.springframework.aot.hint.ReflectionHints;
import org.springframework.aot.hint.RuntimeHints;
import org.springframework.aot.hint.RuntimeHintsRegistrar;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
/**
* {@link RuntimeHintsRegistrar} for Spring Integration MQTT module.
*
* @author Artem Bilan
*
* @since 6.1.9
*/
class MqttRuntimeHints implements RuntimeHintsRegistrar {
@Override
public void registerHints(RuntimeHints hints, ClassLoader classLoader) {
ReflectionHints reflectionHints = hints.reflection();
// TODO until the real fix in Paho library.
Stream.of("org.eclipse.paho.client.mqttv3.MqttAsyncClient", "org.eclipse.paho.mqttv5.client.MqttAsyncClient")
.filter((typeName) -> ClassUtils.isPresent(typeName, classLoader))
.map((typeName) -> loadClassByName(typeName, classLoader))
.flatMap((type) -> Stream.ofNullable(ReflectionUtils.findMethod(type, "stopReconnectCycle")))
.forEach(method -> reflectionHints.registerMethod(method, ExecutableMode.INVOKE));
}
private static Class<?> loadClassByName(String typeName, ClassLoader classLoader) {
try {
return ClassUtils.forName(typeName, classLoader);
}
catch (ClassNotFoundException ex) {
throw new IllegalArgumentException(ex);
}
}
}

6
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/aot/package-info.java Normal file
View File

@@ -0,0 +1,6 @@
/**
* Provides classes to support Spring AOT.
*/
@org.springframework.lang.NonNullApi
@org.springframework.lang.NonNullFields
package org.springframework.integration.mqtt.aot;

6
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv3ClientManager.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022-2023 the original author or authors.
* Copyright 2022-2024 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.
@@ -26,6 +26,7 @@ import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
import org.springframework.integration.mqtt.support.MqttUtils;
import org.springframework.util.Assert;
/**
@@ -149,6 +150,9 @@ public class Mqttv3ClientManager
}
try {
client.disconnectForcibly(getDisconnectCompletionTimeout());
if (getConnectionInfo().isAutomaticReconnect()) {
MqttUtils.stopClientReconnectCycle(client);
}
}
catch (MqttException e) {
logger.error("Could not disconnect from the client", e);

6
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/Mqttv5ClientManager.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022-2023 the original author or authors.
* Copyright 2022-2024 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.
@@ -28,6 +28,7 @@ import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
import org.springframework.integration.mqtt.support.MqttUtils;
import org.springframework.util.Assert;
/**
@@ -151,6 +152,9 @@ public class Mqttv5ClientManager
try {
client.disconnectForcibly(getDisconnectCompletionTimeout());
if (getConnectionInfo().isAutomaticReconnect()) {
MqttUtils.stopClientReconnectCycle(client);
}
}
catch (MqttException e) {
logger.error("Could not disconnect from the client", e);

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

@@ -228,6 +228,9 @@ public class MqttPahoMessageDrivenChannelAdapter
try {
this.client.disconnectForcibly(getDisconnectCompletionTimeout());
if (getConnectionInfo().isAutomaticReconnect()) {
MqttUtils.stopClientReconnectCycle(this.client);
}
}
catch (MqttException ex) {
logger.error(ex, "Exception while disconnecting");

4
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
View File

@@ -53,6 +53,7 @@ import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
import org.springframework.integration.mqtt.support.MqttHeaderMapper;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.integration.mqtt.support.MqttMessageConverter;
import org.springframework.integration.mqtt.support.MqttUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
@@ -299,6 +300,9 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
}
if (getClientManager() == null) {
this.mqttClient.disconnectForcibly(getDisconnectCompletionTimeout());
if (getConnectionInfo().isAutomaticReconnect()) {
MqttUtils.stopClientReconnectCycle(this.mqttClient);
}
}
}
}

5
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/MqttPahoMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2023 the original author or authors.
* Copyright 2002-2024 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.
@@ -176,6 +176,9 @@ public class MqttPahoMessageHandler extends AbstractMqttMessageHandler<IMqttAsyn
IMqttAsyncClient theClient = this.client;
if (theClient != null) {
theClient.disconnect().waitForCompletion(getDisconnectCompletionTimeout());
if (getConnectionInfo().isAutomaticReconnect()) {
MqttUtils.stopClientReconnectCycle(theClient);
}
theClient.close();
this.client = null;
}

4
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java
View File

@@ -41,6 +41,7 @@ import org.springframework.integration.mqtt.event.MqttMessageSentEvent;
import org.springframework.integration.mqtt.event.MqttProtocolErrorEvent;
import org.springframework.integration.mqtt.support.MqttHeaderMapper;
import org.springframework.integration.mqtt.support.MqttMessageConverter;
import org.springframework.integration.mqtt.support.MqttUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
@@ -184,6 +185,9 @@ public class Mqttv5PahoMessageHandler extends AbstractMqttMessageHandler<IMqttAs
try {
if (getClientManager() == null) {
this.mqttClient.disconnect().waitForCompletion(getDisconnectCompletionTimeout());
if (getConnectionInfo().isAutomaticReconnect()) {
MqttUtils.stopClientReconnectCycle(this.mqttClient);
}
}
}
catch (MqttException ex) {

60
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/MqttUtils.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2021 the original author or authors.
* Copyright 2020-2024 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.
@@ -16,9 +16,13 @@
package org.springframework.integration.mqtt.support;
import java.lang.reflect.Method;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.BeanUtils;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
/**
* MQTT Utilities.
@@ -30,6 +34,38 @@ import org.springframework.beans.BeanUtils;
*/
public final class MqttUtils {
private static final boolean PAHO_MQTTV3_PRESENT =
ClassUtils.isPresent("org.eclipse.paho.client.mqttv3.MqttAsyncClient", null);
private static final boolean PAHO_MQTTV5_PRESENT =
ClassUtils.isPresent("org.eclipse.paho.mqttv5.client.MqttAsyncClient", null);
private static final Method V3_STOP_RECONNECT_CYCLE_METHOD;
private static final Method V5_STOP_RECONNECT_CYCLE_METHOD;
static {
if (PAHO_MQTTV3_PRESENT) {
V3_STOP_RECONNECT_CYCLE_METHOD =
ReflectionUtils.findMethod(org.eclipse.paho.client.mqttv3.MqttAsyncClient.class,
"stopReconnectCycle");
ReflectionUtils.makeAccessible(V3_STOP_RECONNECT_CYCLE_METHOD);
}
else {
V3_STOP_RECONNECT_CYCLE_METHOD = null;
}
if (PAHO_MQTTV5_PRESENT) {
V5_STOP_RECONNECT_CYCLE_METHOD =
ReflectionUtils.findMethod(org.eclipse.paho.mqttv5.client.MqttAsyncClient.class,
"stopReconnectCycle");
ReflectionUtils.makeAccessible(V5_STOP_RECONNECT_CYCLE_METHOD);
}
else {
V5_STOP_RECONNECT_CYCLE_METHOD = null;
}
}
private MqttUtils() {
}
@@ -47,4 +83,26 @@ public final class MqttUtils {
return options2;
}
/**
* Perform a {@code stopReconnectCycle()} (via reflection) method on the provided client
* to clean up resources on client stop.
* TODO until the real fix in Paho library.
* @param client the MQTTv3 Paho client instance.
* @since 6.1.9
*/
public static void stopClientReconnectCycle(org.eclipse.paho.client.mqttv3.IMqttAsyncClient client) {
ReflectionUtils.invokeMethod(V3_STOP_RECONNECT_CYCLE_METHOD, client);
}
/**
* Perform a {@code stopReconnectCycle()} (via reflection) method on the provided client
* to clean up resources on client stop.
* TODO until the real fix in Paho library.
* @param client the MQTTv5 Paho client instance.
* @since 6.1.9
*/
public static void stopClientReconnectCycle(org.eclipse.paho.mqttv5.client.IMqttAsyncClient client) {
ReflectionUtils.invokeMethod(V5_STOP_RECONNECT_CYCLE_METHOD, client);
}
}

1
spring-integration-mqtt/src/main/resources/META-INF/spring/aot.factories Normal file
View File

@@ -0,0 +1 @@
org.springframework.aot.hint.RuntimeHintsRegistrar=org.springframework.integration.mqtt.aot.MqttRuntimeHints

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2023 the original author or authors.
* Copyright 2002-2024 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.
@@ -488,8 +488,8 @@ public class MqttAdapterTests {
given(token.getGrantedQos()).willReturn(new int[] {2, 0});
willReturn(token).given(client).subscribe(any(String[].class), any(int[].class), any());
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
"baz", "fix");
MqttPahoMessageDrivenChannelAdapter adapter =
new MqttPahoMessageDrivenChannelAdapter("tcp://mqtt.host", "bar", factory, "baz", "fix");
AtomicReference<Method> method = new AtomicReference<>();
ReflectionUtils.doWithMethods(MqttPahoMessageDrivenChannelAdapter.class, m -> {
m.setAccessible(true);