From 20eb2922ee66393aae0818bf1896692ff22772cc Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Tue, 12 Oct 2010 05:42:10 -0400
Subject: [PATCH] INT-1493, ensured that default 'maxMessagesPerPoll' for SPCA
 is 1 and -1 for PC, added tests validating that both SPCA and PC stops

---
 ...ourcePollingChannelAdapterFactoryBean.java |   3 +
 .../endpoint/AbstractPollingEndpoint.java     |   2 +-
 .../endpoint/PollingLifecycleTests.java       | 116 ++++++++++++++++++
 3 files changed, 120 insertions(+), 1 deletion(-)
 create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingLifecycleTests.java

diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
index 957fbee0a5..74b34789d6 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
@@ -125,6 +125,9 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 				Assert.notNull(this.pollerMetadata, "No poller has been defined for channel-adapter '"
 						+ this.beanName + "', and no default poller is available within the context.");
 			}
+			if (this.pollerMetadata.getMaxMessagesPerPoll() < 1){
+				this.pollerMetadata.setMaxMessagesPerPoll(1);
+			}
 			spca.setPollerMetadata(this.pollerMetadata);
 			
 			spca.setBeanClassLoader(this.beanClassLoader);
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
index 6338a68bc8..6b7c6a120b 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
@@ -47,7 +47,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 
 	private volatile ErrorHandler errorHandler;
 
-	private volatile PollerMetadata pollerMetadata;
+	private volatile PollerMetadata pollerMetadata = new PollerMetadata();
 
 	private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
 
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingLifecycleTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingLifecycleTests.java
new file mode 100644
index 0000000000..dd9c07252a
--- /dev/null
+++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingLifecycleTests.java
@@ -0,0 +1,116 @@
+/*
+ * Copyright 2002-2010 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.endpoint;
+
+import static junit.framework.Assert.assertNotNull;
+import static junit.framework.Assert.assertNull;
+import static junit.framework.Assert.assertTrue;
+import static org.easymock.EasyMock.reset;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessagingException;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
+import org.springframework.integration.config.TestErrorHandler;
+import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.scheduling.support.PeriodicTrigger;
+
+/**
+ * @author Oleg Zhurakousky
+ *
+ */
+public class PollingLifecycleTests {
+	private ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
+	private TestErrorHandler errorHandler = new TestErrorHandler();
+	
+	@Before
+	public void init() throws Exception {
+		taskScheduler.afterPropertiesSet();
+	}
+	
+	@Test
+	public void ensurePollerTaskStops() throws Exception{
+		final CountDownLatch latch = new CountDownLatch(1);
+		QueueChannel channel = new QueueChannel();
+		channel.send(new GenericMessage<String>("foo"));
+
+		MessageHandler handler = Mockito.spy(new MessageHandler() {
+			public void handleMessage(Message<?> message) throws MessagingException {
+				latch.countDown();
+			}
+		});
+		PollingConsumer consumer = new PollingConsumer(channel, handler);
+		PollerMetadata pollerMetadata = new PollerMetadata();
+		pollerMetadata.setTrigger(new PeriodicTrigger(0));
+		consumer.setPollerMetadata(pollerMetadata);
+		consumer.setErrorHandler(errorHandler);
+		consumer.setTaskScheduler(taskScheduler);
+		consumer.setBeanFactory(mock(BeanFactory.class));
+		consumer.afterPropertiesSet();
+		consumer.start();
+		assertTrue(latch.await(2, TimeUnit.SECONDS));
+		consumer.stop();
+		for (int i = 0; i < 10; i++) {
+			channel.send(new GenericMessage<String>("foo"));
+		}
+		Mockito.verify(handler, times(1)).handleMessage(Mockito.any(Message.class));
+	}
+	
+	@Test
+	public void ensurePollerTaskStopsForAdapter() throws Exception{
+		final CountDownLatch latch = new CountDownLatch(1);
+		QueueChannel channel = new QueueChannel();
+		
+		SourcePollingChannelAdapterFactoryBean adapterFactory = new SourcePollingChannelAdapterFactoryBean();
+		PollerMetadata pollerMetadata = new PollerMetadata();
+		pollerMetadata.setMaxMessagesPerPoll(-1); // should be overriden in FB
+		pollerMetadata.setTrigger(new PeriodicTrigger(2000));
+		adapterFactory.setPollerMetadata(pollerMetadata);
+		MessageSource<String> source = spy(new MessageSource<String>() {
+			public Message<String> receive() {
+				latch.countDown();
+				return new GenericMessage<String>("hello");
+			}
+		});
+		adapterFactory.setSource(source);
+		adapterFactory.setOutputChannel(channel);
+		adapterFactory.setBeanFactory(mock(ConfigurableBeanFactory.class));
+		SourcePollingChannelAdapter adapter = adapterFactory.getObject();
+		adapter.setTaskScheduler(taskScheduler);
+		adapter.afterPropertiesSet();
+		adapter.start();
+		assertTrue(latch.await(2, TimeUnit.SECONDS));
+		assertNotNull(channel.receive(100));
+		adapter.stop();
+		assertNull(channel.receive(1000));
+		Mockito.verify(source, times(1)).receive();
+	}
+}