diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index da6bc5ad96..8e1fc735d9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.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. @@ -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 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 b297f4f09e..8eb9af3116 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,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(); 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 63fc95adee..cbf46c13ae 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 @@ -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; }