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**

Conflicts:
	spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
Resolved.
This commit is contained in:
Artem Bilan
2017-11-07 11:44:52 -05:00
committed by Gary Russell
parent 26819f5a95
commit 628d235f7f
2 changed files with 26 additions and 10 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();
}
@@ -313,11 +318,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));
}
}
}

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

@@ -26,15 +26,16 @@ import static org.junit.Assert.assertTrue;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.BDDMockito.willReturn;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyLong;
import static org.mockito.Mockito.anyString;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyLong;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
@@ -56,7 +57,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;
@@ -83,6 +83,8 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 4.0
*
*/
@@ -331,6 +333,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
@@ -372,7 +381,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