INT-4132: Start MS Before Scheduling Polling Task

JIRA: https://jira.spring.io/browse/INT-4132

The race condition is present when polling task may be ran before `MessageSource<?>` has been started.

* Swap the order of `start()` in the `SourcePollingChannelAdapter`.
Since proxying is now applied only for the `MessageSource.receive()` it doesn't hurt to start it before actual proxying.
Just because it is really should be started before performing its `receive()`
* Prove the proper order with the mock test and protect ourselves for the future similar changes
* Also swap the `stop()` order in the `SourcePollingChannelAdapter.
We have to stop/cancel the polling task before discarding internal `MessageSource` lifecycle.
 For example with the current state we may close an underlying resource already, but still have the last polling tick.
 That may cause any unexpected behaviour

 **Cherry-pick to 4.3.x**

Polishing

spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java

@@ -152,19 +152,19 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
 
 	@Override
 	protected void doStart() {
-		super.doStart();
 		if (this.source instanceof Lifecycle) {
 			((Lifecycle) this.source).start();
 		}
+		super.doStart();
 	}
 
 
 	@Override
 	protected void doStop() {
+		super.doStop();
 		if (this.source instanceof Lifecycle) {
 			((Lifecycle) this.source).stop();
 		}
-		super.doStop();
 	}
 
 

spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2016 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,7 +17,10 @@
 package org.springframework.integration.config;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.BDDMockito.willAnswer;
+import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.contains;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -37,12 +40,10 @@ import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
 import org.apache.commons.logging.Log;
 import org.junit.Test;
-import org.mockito.Mockito;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
@@ -54,6 +55,8 @@ import org.springframework.integration.test.util.TestUtils.TestApplicationContex
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.Trigger;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.scheduling.support.PeriodicTrigger;
 import org.springframework.util.ClassUtils;
@@ -72,15 +75,13 @@ public class SourcePollingChannelAdapterFactoryBeanTests {
 		factoryBean.setBeanFactory(context.getBeanFactory());
 		factoryBean.setBeanClassLoader(ClassUtils.getDefaultClassLoader());
 		factoryBean.setOutputChannel(outputChannel);
-		factoryBean.setSource(new TestSource());
+		factoryBean.setSource(() -> new GenericMessage<>("test"));
 		PollerMetadata pollerMetadata = new PollerMetadata();
 		List<Advice> adviceChain = new ArrayList<Advice>();
 		final AtomicBoolean adviceApplied = new AtomicBoolean(false);
-		adviceChain.add(new MethodInterceptor() {
-			public Object invoke(MethodInvocation invocation) throws Throwable {
-				adviceApplied.set(true);
-				return invocation.proceed();
-			}
+		adviceChain.add((MethodInterceptor) invocation -> {
+			adviceApplied.set(true);
+			return invocation.proceed();
 		});
 		pollerMetadata.setTrigger(new PeriodicTrigger(5000));
 		pollerMetadata.setMaxMessagesPerPoll(1);
@@ -95,7 +96,6 @@ public class SourcePollingChannelAdapterFactoryBeanTests {
 		assertTrue("adviceChain was not applied", adviceApplied.get());
 	}
 
-	@SuppressWarnings("rawtypes")
 	@Test
 	public void testTransactionalAdviceChain() throws Throwable {
 		SourcePollingChannelAdapterFactoryBean factoryBean = new SourcePollingChannelAdapterFactoryBean();
@@ -104,31 +104,25 @@ public class SourcePollingChannelAdapterFactoryBeanTests {
 		factoryBean.setBeanFactory(context.getBeanFactory());
 		factoryBean.setBeanClassLoader(ClassUtils.getDefaultClassLoader());
 		factoryBean.setOutputChannel(outputChannel);
-		factoryBean.setSource(new TestSource());
+		factoryBean.setSource(() -> new GenericMessage<>("test"));
 		PollerMetadata pollerMetadata = new PollerMetadata();
 		List<Advice> adviceChain = new ArrayList<Advice>();
 		final AtomicBoolean adviceApplied = new AtomicBoolean(false);
-		adviceChain.add(new MethodInterceptor() {
-			public Object invoke(MethodInvocation invocation) throws Throwable {
-				adviceApplied.set(true);
-				return invocation.proceed();
-			}
+		adviceChain.add((MethodInterceptor) invocation -> {
+			adviceApplied.set(true);
+			return invocation.proceed();
 		});
 		pollerMetadata.setTrigger(new PeriodicTrigger(5000));
 		pollerMetadata.setMaxMessagesPerPoll(1);
 		final AtomicInteger count = new AtomicInteger();
 		final MethodInterceptor txAdvice = mock(MethodInterceptor.class);
-		adviceChain.add(new MethodInterceptor() {
-			public Object invoke(MethodInvocation invocation) throws Throwable {
-				count.incrementAndGet();
-				return invocation.proceed();
-			}
+		adviceChain.add((MethodInterceptor) invocation -> {
+			count.incrementAndGet();
+			return invocation.proceed();
 		});
-		when(txAdvice.invoke(Mockito.any(MethodInvocation.class))).thenAnswer(new Answer() {
-			public Object answer(InvocationOnMock invocation) throws Throwable {
-				count.incrementAndGet();
-				return ((MethodInvocation) invocation.getArguments()[0]).proceed();
-			}
+		when(txAdvice.invoke(any(MethodInvocation.class))).thenAnswer(invocation -> {
+			count.incrementAndGet();
+			return invocation.getArgumentAt(0, MethodInvocation.class).proceed();
 		});
 
 		pollerMetadata.setAdviceChain(adviceChain);
@@ -147,19 +141,16 @@ public class SourcePollingChannelAdapterFactoryBeanTests {
 	public void testInterrupted() throws Exception {
 		final CountDownLatch startLatch = new CountDownLatch(1);
 
-		MessageSource<Object> ms = new MessageSource<Object>() {
-			@Override
-			public Message<Object> receive() {
-				startLatch.countDown();
-				try {
-					Thread.sleep(10000);
-				}
-				catch (InterruptedException e) {
-					Thread.currentThread().interrupt();
-					throw new MessagingException("Interrupted awaiting stopLatch", e);
-				}
-				return null;
+		MessageSource<Object> ms = () -> {
+			startLatch.countDown();
+			try {
+				Thread.sleep(10000);
 			}
+			catch (InterruptedException e) {
+				Thread.currentThread().interrupt();
+				throw new MessagingException("Interrupted awaiting stopLatch", e);
+			}
+			return null;
 		};
 
 		SourcePollingChannelAdapter pollingChannelAdapter = new SourcePollingChannelAdapter();
@@ -199,11 +190,58 @@ public class SourcePollingChannelAdapterFactoryBeanTests {
 		verify(adapterLogger).debug(contains("Poll interrupted - during stop()?"));
 	}
 
-	private static class TestSource implements MessageSource<String> {
+	@Test
+	public void testStartSourceBeforeRunPollingTask() {
+		TaskScheduler taskScheduler = mock(TaskScheduler.class);
 
-		public Message<String> receive() {
-			return new GenericMessage<String>("test");
+		willAnswer(invocation -> {
+			Runnable task = invocation.getArgumentAt(0, Runnable.class);
+			task.run();
+			return null;
+		})
+				.given(taskScheduler)
+				.schedule(any(Runnable.class), any(Trigger.class));
+
+		SourcePollingChannelAdapter pollingChannelAdapter = new SourcePollingChannelAdapter();
+		pollingChannelAdapter.setTaskScheduler(taskScheduler);
+		pollingChannelAdapter.setSource(new LifecycleMessageSource());
+		pollingChannelAdapter.setMaxMessagesPerPoll(1);
+		QueueChannel outputChannel = new QueueChannel();
+		pollingChannelAdapter.setOutputChannel(outputChannel);
+		pollingChannelAdapter.setBeanFactory(mock(BeanFactory.class));
+		pollingChannelAdapter.afterPropertiesSet();
+		pollingChannelAdapter.start();
+
+		Message<?> receive = outputChannel.receive(10_000);
+		assertNotNull(receive);
+		assertEquals(true, receive.getPayload());
+		pollingChannelAdapter.stop();
+	}
+
+	private static class LifecycleMessageSource implements MessageSource<Boolean>, Lifecycle {
+
+		private volatile boolean running;
+
+		@Override
+		public void start() {
+			this.running = true;
 		}
+
+		@Override
+		public void stop() {
+			this.running = false;
+		}
+
+		@Override
+		public boolean isRunning() {
+			return this.running;
+		}
+
+		@Override
+		public Message<Boolean> receive() {
+			return new GenericMessage<>(isRunning());
+		}
+
 	}
 
 }