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`.

Conflicts:
	spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
	spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java

* Fix single-arg `Assert` method usage in the `AbstractCorrelatingMessageHandler`
This commit is contained in:
Gary Russell
2017-02-02 12:10:44 -05:00
committed by Artem Bilan
parent 6e7653f18f
commit 4c68e22d87
3 changed files with 57 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.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.
@@ -132,9 +132,9 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
Assert.notNull(processor);
Assert.notNull(processor, "'processor' must not be null");
Assert.notNull(store, "'store' must not be null");
Assert.notNull(store);
setMessageStore(store);
this.outputProcessor = processor;
this.correlationStrategy = (correlationStrategy == null

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,7 +273,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
private void scheduleReconnect() {
try {
this.reconnectFuture = this.getTaskScheduler().scheduleWithFixedDelay(new Runnable() {
this.reconnectFuture = getTaskScheduler().schedule(new Runnable() {
@Override
public void run() {
@@ -283,16 +281,20 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
if (logger.isDebugEnabled()) {
logger.debug("Attempting reconnect");
}
if (!MqttPahoMessageDrivenChannelAdapter.this.connected) {
connectAndSubscribe();
synchronized (MqttPahoMessageDrivenChannelAdapter.this) {
if (!MqttPahoMessageDrivenChannelAdapter.this.connected) {
MqttPahoMessageDrivenChannelAdapter.this.connectAndSubscribe();
MqttPahoMessageDrivenChannelAdapter.this.reconnectFuture = null;
}
}
}
catch (MqttException e) {
logger.error("Exception while connecting and subscribing", e);
MqttPahoMessageDrivenChannelAdapter.this.scheduleReconnect();
}
}
}, this.recoveryInterval);
}, new Date(System.currentTimeMillis() + this.recoveryInterval));
}
catch (Exception e) {
logger.error("Failed to schedule reconnect", e);
@@ -300,7 +302,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();

50
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.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,11 +17,14 @@
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;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doAnswer;
@@ -39,12 +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.aopalliance.intercept.MethodInvocation;
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;
@@ -60,6 +64,7 @@ import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
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.integration.channel.NullChannel;
@@ -72,8 +77,10 @@ import org.springframework.integration.mqtt.event.MqttIntegrationEvent;
import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
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;
/**
@@ -87,14 +94,7 @@ public class MqttAdapterTests {
{
ProxyFactoryBean pfb = new ProxyFactoryBean();
pfb.addAdvice(new MethodInterceptor() {
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
return null;
}
});
pfb.addAdvice((MethodInterceptor) invocation -> null);
pfb.setInterfaces(IMqttToken.class);
this.alwaysComplete = (IMqttToken) pfb.getObject();
}
@@ -336,6 +336,7 @@ public class MqttAdapterTests {
}
assertThat(event, instanceOf(MqttSubscribedEvent.class));
assertEquals("Connected and subscribed to [baz, fix]", ((MqttSubscribedEvent) event).getMessage());
taskScheduler.destroy();
}
@Test
@@ -379,6 +380,34 @@ public class MqttAdapterTests {
verifyNotUnsubscribe(client);
}
@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() {
@@ -404,6 +433,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;
}