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

INT-3923: Fix MQTT Reconnect Logic

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3923

Fixes: #2046

Previously, when connection is lost, the inbound adapter attempted to reconnect on
a schedule with a fixed delay.

If a connection was again lost, while the schedule is still running, we can end up
with another scheduled task running.

This is benign aside from the DEBUG log noise because the scheduled task tests the
connection before reconnecting.

However, if the `recoveryInterval` is short, it could consume CPU.

Change the reconnect to be a one-time scheduled task and reschedule if it fails to
reconnect.

Synchronize all access to the `connected` field.

Add a test case with a short recovery interval, before this fix, we see many logs
`Attempting reconnect`.
This commit is contained in:
Gary Russell
2017-02-02 12:10:44 -05:00
committed by Artem Bilan
parent 218daa94e8
commit d8cbcd1310
2 changed files with 51 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -17,6 +17,7 @@
package org.springframework.integration.mqtt.inbound;
import java.util.Arrays;
import java.util.Date;
import java.util.concurrent.ScheduledFuture;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
@@ -138,6 +139,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
@Override
protected void doStart() {
Assert.state(getTaskScheduler() != null, "A 'taskScheduler' is required");
super.doStart();
try {
connectAndSubscribe();
@@ -149,7 +151,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
}
@Override
protected void doStop() {
protected synchronized void doStop() {
cancelReconnect();
super.doStop();
if (this.client != null) {
@@ -219,7 +221,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
}
}
private void connectAndSubscribe() throws MqttException {
private synchronized void connectAndSubscribe() throws MqttException {
MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions();
this.cleanSession = connectionOptions.isCleanSession();
this.consumerStopAction = this.clientFactory.getConsumerStopAction();
@@ -259,10 +261,6 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(new MqttSubscribedEvent(this, message));
}
// cancel() after the publish in case we are on that thread; a send to a QueueChannel would fail.
if (this.reconnectFuture != null) {
cancelReconnect();
}
}
}
@@ -275,19 +273,23 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
private void scheduleReconnect() {
try {
this.reconnectFuture = this.getTaskScheduler().scheduleWithFixedDelay(() -> {
this.reconnectFuture = getTaskScheduler().schedule(() -> {
try {
if (logger.isDebugEnabled()) {
logger.debug("Attempting reconnect");
}
if (!MqttPahoMessageDrivenChannelAdapter.this.connected) {
connectAndSubscribe();
synchronized (MqttPahoMessageDrivenChannelAdapter.this) {
if (!MqttPahoMessageDrivenChannelAdapter.this.connected) {
connectAndSubscribe();
MqttPahoMessageDrivenChannelAdapter.this.reconnectFuture = null;
}
}
}
catch (MqttException e) {
logger.error("Exception while connecting and subscribing", e);
scheduleReconnect();
}
}, this.recoveryInterval);
}, new Date(System.currentTimeMillis() + this.recoveryInterval));
}
catch (Exception e) {
logger.error("Failed to schedule reconnect", e);
@@ -295,7 +297,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
}
@Override
public void connectionLost(Throwable cause) {
public synchronized void connectionLost(Throwable cause) {
this.logger.error("Lost connection:" + cause.getMessage() + "; retrying...");
this.connected = false;
scheduleReconnect();

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

@@ -17,6 +17,7 @@
package org.springframework.integration.mqtt;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.lessThanOrEqualTo;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
@@ -24,6 +25,8 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
@@ -39,11 +42,13 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import javax.net.SocketFactory;
import org.aopalliance.intercept.MethodInterceptor;
import org.apache.commons.logging.Log;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
@@ -57,6 +62,7 @@ import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.Test;
import org.springframework.aop.framework.ProxyFactoryBean;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -77,6 +83,7 @@ import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
@@ -301,6 +308,7 @@ public class MqttAdapterTests {
}
assertThat(event, instanceOf(MqttSubscribedEvent.class));
assertEquals("Connected and subscribed to [baz, fix]", ((MqttSubscribedEvent) event).getMessage());
taskScheduler.destroy();
}
@Test
@@ -369,6 +377,34 @@ public class MqttAdapterTests {
ctx.close();
}
@Test
public void testReconnect() throws Exception {
final MqttAsyncClient client = mock(MqttAsyncClient.class);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
adapter.setRecoveryInterval(10);
Log logger = spy(TestUtils.getPropertyValue(adapter, "logger", Log.class));
new DirectFieldAccessor(adapter).setPropertyValue("logger", logger);
given(logger.isDebugEnabled()).willReturn(true);
final AtomicInteger attemptingReconnectCount = new AtomicInteger();
willAnswer(i -> {
if (attemptingReconnectCount.getAndIncrement() == 0) {
adapter.connectionLost(new RuntimeException("while schedule running"));
}
i.callRealMethod();
return null;
}).given(logger).debug("Attempting reconnect");
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
adapter.setTaskScheduler(taskScheduler);
adapter.start();
adapter.connectionLost(new RuntimeException("initial"));
Thread.sleep(1000);
// the following assertion should be equalTo, but leq to protect against a slow CI server
assertThat(attemptingReconnectCount.get(), lessThanOrEqualTo(2));
adapter.stop();
taskScheduler.destroy();
}
private MqttPahoMessageDrivenChannelAdapter buildAdapter(final MqttAsyncClient client, Boolean cleanSession,
ConsumerStopAction action) throws MqttException, MqttSecurityException {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() {
@@ -394,6 +430,7 @@ public class MqttAdapterTests {
MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("client", factory, "foo");
adapter.setApplicationEventPublisher(mock(ApplicationEventPublisher.class));
adapter.setOutputChannel(new NullChannel());
adapter.setTaskScheduler(mock(TaskScheduler.class));
adapter.afterPropertiesSet();
return adapter;
}