diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateAsynchronousTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateAsynchronousTests.java index e55d8fc3f..e8a1c9541 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateAsynchronousTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/repeat/support/TaskExecutorRepeatTemplateAsynchronousTests.java @@ -23,11 +23,14 @@ import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; +import java.util.List; import java.util.Set; import org.junit.Test; +import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.repeat.RepeatCallback; import org.springframework.batch.repeat.RepeatContext; import org.springframework.batch.repeat.RepeatStatus; @@ -124,7 +127,7 @@ public class TaskExecutorRepeatTemplateAsynchronousTests extends AbstractTradeBa return super.doInIteration(context); } }); - + assertTrue("Too few attempts: " + count, count >= 1); assertTrue("Too many attempts: " + count, count <= 10); @@ -165,22 +168,35 @@ public class TaskExecutorRepeatTemplateAsynchronousTests extends AbstractTradeBa @Test public void testThrottleLimit() throws Exception { + + int throttleLimit = 600; + TaskExecutorRepeatTemplate template = new TaskExecutorRepeatTemplate(); SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); - taskExecutor.setConcurrencyLimit(3); + taskExecutor.setConcurrencyLimit(300); template.setTaskExecutor(taskExecutor); - template.setThrottleLimit(12); + template.setThrottleLimit(throttleLimit); final String threadName = Thread.currentThread().getName(); final Set threadNames = new HashSet(); + final List items = new ArrayList(); final RepeatCallback callback = new RepeatCallback() { public RepeatStatus doInIteration(RepeatContext context) throws Exception { assertNotSame(threadName, Thread.currentThread().getName()); - threadNames.add(Thread.currentThread().getName()); Trade item = provider.read(); + threadNames.add(Thread.currentThread().getName() + " : " + item); + items.add("" + item); if (item != null) { processor.write(Collections.singletonList(item)); + // Do some more I/O + for (int i = 0; i < 10; i++) { + TradeItemReader provider = new TradeItemReader(resource); + provider.open(new ExecutionContext()); + while (provider.read() != null) + continue; + provider.close(); + } } return RepeatStatus.continueIf(item != null); } @@ -191,6 +207,10 @@ public class TaskExecutorRepeatTemplateAsynchronousTests extends AbstractTradeBa // Thread.sleep(500); assertEquals(NUMBER_OF_ITEMS, processor.count); assertTrue(threadNames.size() > 1); + int frequency = Collections.frequency(items, "null"); + // System.err.println("Frequency: "+frequency); + assertTrue(frequency <= throttleLimit); + assertTrue(frequency > 1); } /** @@ -215,6 +235,10 @@ public class TaskExecutorRepeatTemplateAsynchronousTests extends AbstractTradeBa assertNotSame(threadName, Thread.currentThread().getName()); threadNames.add(Thread.currentThread().getName()); Thread.sleep(100); + TradeItemReader provider = new TradeItemReader(resource); + provider.open(new ExecutionContext()); + while (provider.read() != null) + ; return super.doInIteration(context); } };