Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1853 changed default max-messages-per-poll to be Integer.MIN_VALUE

Browse Source
This commit is contained in:
Oleg Zhurakousky
2011-05-03 16:52:53 -04:00
parent 2150b9b806
commit def3647c80
4 changed files with 46 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-core/.springBeans
View File

@@ -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>

2
spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
View File

@@ -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);

2
spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java
View File

@@ -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;

42
spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingLifecycleTests.java
View File

@@ -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();
}
}