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

GH-3955: Mqtt adapter unsubscribe when cleanStart

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

`Mqttv5PahoMessageDrivenChannelAdapter` unsubscribes from all topics, even if `cleanStart/cleanSession` is set to `false`, thus not receiving offline messages after restart.

* unsubscribe `Mqttv5PahoMessageDrivenChannelAdapter` only when `cleanStart`
* add tests

**Cherry-pick to `5.5.x`**
# Conflicts:
#	spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
This commit is contained in:
mths1
2022-12-01 11:36:23 -05:00
committed by abilan
parent 067de96ab3
commit 36e4fe1fb5
2 changed files with 105 additions and 1 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

@@ -67,6 +67,7 @@ import org.springframework.util.Assert;
* @author Artem Bilan
* @author Mikhail Polivakha
* @author Lucas Bowler
* @author Matthias Thoma
*
* @since 5.5.5
*
@@ -190,7 +191,9 @@ public class Mqttv5PahoMessageDrivenChannelAdapter extends AbstractMqttMessageDr
String[] topics = getTopic();
try {
if (this.mqttClient != null && this.mqttClient.isConnected()) {
this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
if (this.connectionOptions.isCleanStart()) {
this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
}
this.mqttClient.disconnect().waitForCompletion(getCompletionTimeout());
}
}

101
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5AdapterTests.java Normal file
View File

@@ -0,0 +1,101 @@
/*
* Copyright 2022 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;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
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.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttSubscription;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.mqtt.inbound.Mqttv5PahoMessageDrivenChannelAdapter;
import org.springframework.test.util.ReflectionTestUtils;
/**
* @author Gary Russell
* @author Artem Bilan
* @author Matthias Thoma
*
* @since 5.5.16
*
*/
public class Mqttv5AdapterTests {
@Test
public void testStop() throws Exception {
final IMqttAsyncClient client = mock(IMqttAsyncClient.class);
Mqttv5PahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, true);
adapter.start();
adapter.connectComplete(false, null);
adapter.stop();
verify(client).connect(any(MqttConnectionOptions.class));
verify(client).subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener[].class), any());
verify(client).unsubscribe(any(String[].class));
}
@Test
public void testStopNotClean() throws Exception {
final IMqttAsyncClient client = mock(IMqttAsyncClient.class);
Mqttv5PahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, false);
adapter.start();
adapter.connectComplete(false, null);
adapter.stop();
verify(client).connect(any(MqttConnectionOptions.class));
verify(client).subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener[].class), any());
verify(client, never()).unsubscribe(any(String[].class));
}
private static Mqttv5PahoMessageDrivenChannelAdapter buildAdapterIn(IMqttAsyncClient client,
boolean cleanStart) throws MqttException {
MqttConnectionOptions connectionOptions = new MqttConnectionOptions();
connectionOptions.setServerURIs(new String[] {"tcp://localhost:1883"});
connectionOptions.setCleanStart(cleanStart);
given(client.isConnected()).willReturn(true);
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.unsubscribe(any(String[].class))).willReturn(token);
Mqttv5PahoMessageDrivenChannelAdapter adapter =
new Mqttv5PahoMessageDrivenChannelAdapter(connectionOptions, "client", "foo");
ReflectionTestUtils.setField(adapter, "mqttClient", client);
adapter.setBeanFactory(mock(BeanFactory.class));
adapter.setApplicationEventPublisher(mock(ApplicationEventPublisher.class));
adapter.setOutputChannel(new NullChannel());
adapter.afterPropertiesSet();
return adapter;
}
}