From def3647c80bb886e7fdc2d3e692ff763b75f0dcd Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@vmware.com>
Date: Tue, 3 May 2011 16:52:53 -0400
Subject: [PATCH] INT-1853 changed default max-messages-per-poll to be
 Integer.MIN_VALUE

---
 spring-integration-core/.springBeans          |  3 +-
 ...ourcePollingChannelAdapterFactoryBean.java |  2 +-
 .../scheduling/PollerMetadata.java            |  2 +-
 .../endpoint/PollingLifecycleTests.java       | 42 +++++++++++++++++++
 4 files changed, 46 insertions(+), 3 deletions(-)

diff --git a/spring-integration-core/.springBeans b/spring-integration-core/.springBeans
index 5505ed4aa2..041e617c24 100644
--- a/spring-integration-core/.springBeans
+++ b/spring-integration-core/.springBeans
@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <beansProjectDescription>
 	<version>1</version>
-	<pluginVersion><![CDATA[2.6.0.201102250324-M2]]></pluginVersion>
+	<pluginVersion><![CDATA[2.6.0.201103160035-RELEASE]]></pluginVersion>
 	<configSuffixes>
 		<configSuffix><![CDATA[xml]]></configSuffix>
 	</configSuffixes>
@@ -10,6 +10,7 @@
 		<config>src/test/java/org/springframework/integration/gateway/GatewayInterfaceTest-context.xml</config>
 		<config>src/test/java/org/springframework/integration/gateway/InnerGatewayWithChainTests-context.xml</config>
 		<config>src/test/java/org/springframework/integration/history/annotated-config.xml</config>
+		<config>src/test/java/vfabric/uk/si-config.xml</config>
 	</configs>
 	<configSets>
 	</configSets>
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 0eb8304df0..9435800762 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
@@ -130,7 +130,7 @@ 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){
+			if (this.pollerMetadata.getMaxMessagesPerPoll() == Integer.MIN_VALUE){
 				// the default is 1 since a source might return
 				// a non-null and non-interruptable value every time it is invoked
 				this.pollerMetadata.setMaxMessagesPerPoll(1);
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java
index bf85e50d4b..affb5fedf1 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java
@@ -30,7 +30,7 @@ import org.springframework.util.ErrorHandler;
  */
 public class PollerMetadata {
 
-	public static final int MAX_MESSAGES_UNBOUNDED = -1;
+	public static final int MAX_MESSAGES_UNBOUNDED = Integer.MIN_VALUE;
 	
 	private volatile Trigger trigger;
 
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
index b770bd931e..350805fd8a 100644
--- 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
@@ -115,4 +115,46 @@ public class PollingLifecycleTests {
 		assertNull(channel.receive(1000));
 		Mockito.verify(source, times(1)).receive();
 	}
+	
+	@Test
+	public void ensurePollerTaskStopsForAdapterWithInteraptable() throws Exception{
+		final CountDownLatch latch = new CountDownLatch(2);
+		QueueChannel channel = new QueueChannel();
+		
+		SourcePollingChannelAdapterFactoryBean adapterFactory = new SourcePollingChannelAdapterFactoryBean();
+		PollerMetadata pollerMetadata = new PollerMetadata();
+		pollerMetadata.setMaxMessagesPerPoll(-1);
+		pollerMetadata.setTrigger(new PeriodicTrigger(2000));
+		adapterFactory.setPollerMetadata(pollerMetadata);
+		final Runnable runnable = mock(Runnable.class);
+		final Runnable coughtInterrupted = mock(Runnable.class);
+		MessageSource<String> source = new MessageSource<String>() {
+			public Message<String> receive() {
+				
+				try {
+					for (int i = 0; i < 10; i++) {
+						runnable.run();
+						latch.countDown();
+						Thread.sleep(1000);
+					}
+				} catch (InterruptedException e) {
+					coughtInterrupted.run();
+				}
+				
+				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(3000, TimeUnit.SECONDS));
+		//
+		adapter.stop();
+		Mockito.verify(runnable, times(2)).run();
+		Mockito.verify(coughtInterrupted, times(1)).run();
+	}
 }