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

GH-8873: Fix on-demand subscription for MQTT v5

Browse Source 
Fixes: #8873

The `mqttClient.subscribe()` API does not check if properties are provided and fails on the
`subscriptionProperties.getSubscriptionIdentifiers().get(0)` call with an `IndexOutOfBoundsException`

* Use another `mqttClient.subscribe()` API in the `Mqttv5PahoMessageDrivenChannelAdapter` where there is not such a check
* Ensure that `addTopic(NAME)` works as expected in the `Mqttv5BackToBackTests`

**Cherry-pick to `6.2.x` & `6.1.x`**
This commit is contained in:
Artem Bilan
2024-01-30 17:03:24 -05:00
parent 25874b3ab9
commit d3924db5a1
2 changed files with 20 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -336,7 +336,8 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
this.subscriptions.add(subscription);
}
if (this.mqttClient != null && this.mqttClient.isConnected()) {
this.mqttClient.subscribe(subscription, this::messageArrived)
this.mqttClient.subscribe(new MqttSubscription[] { subscription },
null, null, new IMqttMessageListener[] { this::messageArrived }, new MqttProperties())
.waitForCompletion(getCompletionTimeout());
}
}
@@ -466,7 +467,7 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
IMqttMessageListener[] listeners = IntStream.range(0, mqttSubscriptions.length)
.mapToObj(t -> listener)
.toArray(IMqttMessageListener[]::new);
this.mqttClient.subscribe(mqttSubscriptions, null, null, listeners, null)
this.mqttClient.subscribe(mqttSubscriptions, null, null, listeners, new MqttProperties())
.waitForCompletion(getCompletionTimeout());
String message = "Connected and subscribed to " + Arrays.toString(mqttSubscriptions);
logger.debug(message);

17
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackTests.java
View File

@@ -75,6 +75,9 @@ public class Mqttv5BackToBackTests implements MosquittoContainerTest {
@Autowired
private Config config;
@Autowired
private Mqttv5PahoMessageDrivenChannelAdapter mqttv5MessageDrivenChannelAdapter;
@Test //GH-3732
public void testNoNpeIsNotThrownInCaseDoInitIsNotInvokedBeforeTopicAddition() {
Mqttv5PahoMessageDrivenChannelAdapter channelAdapter =
@@ -118,6 +121,20 @@ public class Mqttv5BackToBackTests implements MosquittoContainerTest {
.hasAtLeastOneElementOfType(MqttMessageSentEvent.class)
.hasAtLeastOneElementOfType(MqttMessageDeliveredEvent.class)
.hasAtLeastOneElementOfType(MqttSubscribedEvent.class);
this.mqttv5MessageDrivenChannelAdapter.addTopic("anotherTopic");
testPayload = "another payload";
this.mqttOutFlowInput.send(
MessageBuilder.withPayload(testPayload)
.setHeader(MqttHeaders.TOPIC, "anotherTopic")
.build());
receive = this.fromMqttChannel.receive(10_000);
assertThat(receive).isNotNull();
assertThat(receive.getPayload()).isEqualTo(testPayload);
}