diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java index 00be38c522..07f3beedbb 100644 --- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java +++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java @@ -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(); diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java index a8b31db709..c4af79a4ee 100644 --- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java +++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java @@ -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; }