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

GH-9597: Fix Mqttv5PahoMessageHandler.publish() for concurrency

Browse Source 
Fixes: #9597
Issue link: https://github.com/spring-projects/spring-integration/issues/9597

When input channel for the endpoint with a `Mqttv5PahoMessageHandler` is an `ExecutorChannel` and a lot of concurrent messages are sent,
the `Connect already in progress (32110)` exception can be thrown

* Wrap `this.mqttClient.connect()` in the `Mqttv5PahoMessageHandler.publish()` into a `Lock`

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
This commit is contained in:
Artem Bilan
2024-10-25 16:52:27 -04:00
parent 91494094ac
commit 334620f36e
1 changed files with 9 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -241,7 +241,15 @@ public class Mqttv5PahoMessageHandler extends AbstractMqttMessageHandler<IMqttAs
long completionTimeout = getCompletionTimeout();
try {
if (!this.mqttClient.isConnected()) {
this.mqttClient.connect(this.connectionOptions).waitForCompletion(completionTimeout);
this.lock.lock();
try {
if (!this.mqttClient.isConnected()) {
this.mqttClient.connect(this.connectionOptions).waitForCompletion(completionTimeout);
}
}
finally {
this.lock.unlock();
}
}
IMqttToken token =
this.mqttClient.publish(topic, (MqttMessage) mqttMessage, null, this.mqttPublishActionListener);