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

GH-9214: Fix subscription identifier logic in Mqttv5PahoMessageDrivenChannelAdapter

Browse Source 
Fixes: #9214

The `MqttAsyncClient` can set `subscriptionIdentifier` from session if it is enabled
and available from connection

* Remove manual `subscriptionIdentifierCounter` from the `Mqttv5PahoMessageDrivenChannelAdapter`.
Instead, use `subscriptionProperties.setSubscriptionIdentifiers(List.of(0));`
to make the `this.mqttSession.getNextSubscriptionIdentifier();` to work
when `if (connOpts.useSubscriptionIdentifiers() && this.mqttConnection.isSubscriptionIdentifiersAvailable()) {` is `true`

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
This commit is contained in:
Artem Bilan
2024-06-06 11:54:26 -04:00
parent e9561b41f1
commit 3dcbdef330
2 changed files with 14 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -21,13 +21,11 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.IntStream;
import org.eclipse.paho.mqttv5.client.IMqttAsyncClient;
import org.eclipse.paho.mqttv5.client.IMqttMessageListener;
import org.eclipse.paho.mqttv5.client.IMqttToken;
import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttCallback;
@@ -109,8 +107,6 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
private volatile boolean readyToSubscribeOnStart;
private final AtomicInteger subscriptionIdentifierCounter = new AtomicInteger(0);
/**
* Create an instance based on the MQTT url, client id and subscriptions.
* @param url the MQTT url to connect.
@@ -344,9 +340,10 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
}
if (this.mqttClient != null && this.mqttClient.isConnected()) {
MqttProperties subscriptionProperties = new MqttProperties();
subscriptionProperties.setSubscriptionIdentifier(this.subscriptionIdentifierCounter.incrementAndGet());
// Make use of mqttSession.getNextSubscriptionIdentifier() if available in connection
subscriptionProperties.setSubscriptionIdentifiers(List.of(0));
this.mqttClient.subscribe(new MqttSubscription[] {subscription},
null, null, new IMqttMessageListener[] {this::messageArrived}, subscriptionProperties)
null, null, this::messageArrived, subscriptionProperties)
.waitForCompletion(getCompletionTimeout());
}
}
@@ -472,15 +469,10 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher();
this.topicLock.lock();
try {
IMqttMessageListener listener = this::messageArrived;
IMqttMessageListener[] listeners = IntStream.range(0, mqttSubscriptions.length)
.mapToObj(t -> listener)
.toArray(IMqttMessageListener[]::new);
MqttProperties subscriptionProperties = new MqttProperties();
subscriptionProperties.setSubscriptionIdentifiers(IntStream.range(0, mqttSubscriptions.length)
.mapToObj(i -> this.subscriptionIdentifierCounter.incrementAndGet())
.toList());
this.mqttClient.subscribe(mqttSubscriptions, null, null, listeners, subscriptionProperties)
// Make use of mqttSession.getNextSubscriptionIdentifier() if available in connection
subscriptionProperties.setSubscriptionIdentifiers(List.of(0));
this.mqttClient.subscribe(mqttSubscriptions, null, null, this::messageArrived, subscriptionProperties)
.waitForCompletion(getCompletionTimeout());
String message = "Connected and subscribed to " + Arrays.toString(mqttSubscriptions);
logger.debug(message);

13
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5AdapterTests.java
View File

@@ -57,7 +57,7 @@ public class Mqttv5AdapterTests {
adapter.stop();
verify(client).connect(any(MqttConnectionOptions.class));
verify(client).subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener[].class), any());
verify(client).subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener.class), any());
verify(client).unsubscribe(any(String[].class));
}
@@ -71,11 +71,12 @@ public class Mqttv5AdapterTests {
adapter.stop();
verify(client).connect(any(MqttConnectionOptions.class));
verify(client).subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener[].class), any());
verify(client).subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener.class), any());
verify(client, never()).unsubscribe(any(String[].class));
}
private static Mqttv5PahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttAsyncClient client, boolean cleanStart) throws MqttException {
private static Mqttv5PahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttAsyncClient client,
boolean cleanStart) throws MqttException {
MqttConnectionOptions connectionOptions = new MqttConnectionOptions();
connectionOptions.setServerURIs(new String[] {"tcp://localhost:1883"});
@@ -85,9 +86,11 @@ public class Mqttv5AdapterTests {
IMqttToken token = mock(IMqttToken.class);
given(client.disconnect()).willReturn(token);
given(client.connect(any(MqttConnectionOptions.class))).willReturn(token);
given(client.subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener[].class), any())).willReturn(token);
given(client.subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener.class), any()))
.willReturn(token);
given(client.unsubscribe(any(String[].class))).willReturn(token);
Mqttv5PahoMessageDrivenChannelAdapter adapter = new Mqttv5PahoMessageDrivenChannelAdapter(connectionOptions, "client", "foo");
Mqttv5PahoMessageDrivenChannelAdapter adapter =
new Mqttv5PahoMessageDrivenChannelAdapter(connectionOptions, "client", "foo");
ReflectionTestUtils.setField(adapter, "mqttClient", client);
adapter.setBeanFactory(mock(BeanFactory.class));
adapter.setApplicationEventPublisher(mock(ApplicationEventPublisher.class));