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

GH-2275: Fix MqttMDCA for callbacks after stop

Browse Source 
Resolves: spring-projects/spring-integration#2275

* Remove the `MqttPahoMessageDrivenChannelAdapter` as a callback from
the `IMqttClient` during `stop()`
* Check the `isRunning()` from the `connectionLost` callback to avoid
unexpected `scheduleReconnect()` when we are not running

**Cherry-pick to 4.3.x**
This commit is contained in:
Artem Bilan
2017-11-07 11:44:52 -05:00
committed by Gary Russell
parent 098e8bc58f
commit 23137fd0cb
2 changed files with 22 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -43,6 +43,8 @@ import org.springframework.util.Assert;
* Eclipse Paho Implementation.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 1.0
*
*/
@@ -172,6 +174,9 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
catch (MqttException e) {
logger.error("Exception while disconnecting", e);
}
this.client.setCallback(null);
try {
this.client.close();
}
@@ -308,11 +313,13 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
@Override
public synchronized void connectionLost(Throwable cause) {
this.logger.error("Lost connection:" + cause.getMessage() + "; retrying...");
this.connected = false;
scheduleReconnect();
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, cause));
if (isRunning()) {
this.logger.error("Lost connection: " + cause.getMessage() + "; retrying...");
this.connected = false;
scheduleReconnect();
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, cause));
}
}
}

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

@@ -40,6 +40,7 @@ import static org.mockito.Mockito.verify;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
@@ -62,7 +63,6 @@ import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.MqttToken;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.Test;
@@ -97,6 +97,7 @@ import org.springframework.util.ReflectionUtils;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 4.0
*
*/
@@ -345,6 +346,13 @@ public class MqttAdapterTests {
adapter.start();
adapter.stop();
verifyUnsubscribe(client);
adapter.connectionLost(new RuntimeException("Intentional"));
TaskScheduler taskScheduler = TestUtils.getPropertyValue(adapter, "taskScheduler", TaskScheduler.class);
verify(taskScheduler, never())
.schedule(any(Runnable.class), any(Date.class));
}
@Test
@@ -509,7 +517,7 @@ public class MqttAdapterTests {
}
private MqttPahoMessageDrivenChannelAdapter buildAdapter(final IMqttClient client, Boolean cleanSession,
ConsumerStopAction action) throws MqttException, MqttSecurityException {
ConsumerStopAction action) throws MqttException {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() {
@Override