From 130f72a1d8c3c0c7ff1d58fcc2e251f4b66dfee5 Mon Sep 17 00:00:00 2001 From: robokaso Date: Mon, 10 Mar 2008 11:57:47 +0000 Subject: [PATCH] BATCH-410: fixed chunk listener registration and documented current unintuitive behavior in test --- .../core/step/DefaultStepFactoryBean.java | 6 +- .../step/DefaultStepFactoryBeanTests.java | 64 +++++++++++++++---- 2 files changed, 54 insertions(+), 16 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/DefaultStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/DefaultStepFactoryBean.java index b06270515..69d941783 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/DefaultStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/DefaultStepFactoryBean.java @@ -192,17 +192,19 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean { BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper(); + stepOperations = new RepeatTemplate(); + if (commitInterval > 0) { RepeatTemplate chunkOperations = new RepeatTemplate(); chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(commitInterval)); - helper.addChunkListeners(chunkOperations, listeners); + helper.addChunkListeners(stepOperations, listeners); step.setChunkOperations(chunkOperations); } StepListener[] stepListeners = helper.getStepListeners(listeners); itemReader = helper.getItemReader(itemReader, listeners); itemWriter = helper.getItemWriter(itemWriter, listeners); - stepOperations = new RepeatTemplate(); + // In case they are used by subclasses: setItemReader(itemReader); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/DefaultStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/DefaultStepFactoryBeanTests.java index 274d47bf2..efd30bc5a 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/DefaultStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/DefaultStepFactoryBeanTests.java @@ -25,6 +25,7 @@ import junit.framework.TestCase; import org.springframework.batch.core.BatchListener; import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.job.SimpleJob; @@ -33,9 +34,6 @@ import org.springframework.batch.core.repository.support.SimpleJobRepository; import org.springframework.batch.core.repository.support.dao.MapJobExecutionDao; import org.springframework.batch.core.repository.support.dao.MapJobInstanceDao; import org.springframework.batch.core.repository.support.dao.MapStepExecutionDao; -import org.springframework.batch.core.step.AbstractStep; -import org.springframework.batch.core.step.DefaultStepFactoryBean; -import org.springframework.batch.core.step.ItemOrientedStep; import org.springframework.batch.item.AbstractItemWriter; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; @@ -53,15 +51,15 @@ public class DefaultStepFactoryBeanTests extends TestCase { private SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(), new MapStepExecutionDao()); - private List processed = new ArrayList(); + private List written = new ArrayList(); - private ItemWriter processor = new AbstractItemWriter() { + private ItemWriter writer = new AbstractItemWriter() { public void write(Object data) throws Exception { - processed.add((String) data); + written.add((String) data); } }; - private ItemReader provider; + private ItemReader reader; private SimpleJob job = new SimpleJob();; @@ -86,10 +84,10 @@ public class DefaultStepFactoryBeanTests extends TestCase { List items = TransactionAwareProxyFactory.createTransactionalList(); items.addAll(Arrays.asList(args)); - provider = new ListItemReader(items); + reader = new ListItemReader(items); - factory.setItemReader(provider); - factory.setItemWriter(processor); + factory.setItemReader(reader); + factory.setItemWriter(writer); factory.setJobRepository(repository); factory.setTransactionManager(new ResourcelessTransactionManager()); factory.setBeanName("stepName"); @@ -110,8 +108,8 @@ public class DefaultStepFactoryBeanTests extends TestCase { job.execute(jobExecution); assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); - assertEquals(3, processed.size()); - assertTrue(processed.contains("foo")); + assertEquals(3, written.size()); + assertTrue(written.contains("foo")); } public void testSimpleJobWithItemListeners() throws Exception { @@ -158,9 +156,9 @@ public class DefaultStepFactoryBeanTests extends TestCase { job.execute(jobExecution); assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); - assertEquals(0, processed.size()); + assertEquals(0, written.size()); // provider should be exhausted - assertEquals(null, provider.read()); + assertEquals(null, reader.read()); assertEquals(3, recovered.size()); } @@ -186,5 +184,43 @@ public class DefaultStepFactoryBeanTests extends TestCase { } assertEquals(BatchStatus.FAILED, jobExecution.getStatus()); } + + public void testChunkListeners() throws Exception { + String[] items = new String[] { "1", "2", "3", "4", "5", "6", "7" }; + int commitInterval = 3; + + DefaultStepFactoryBean factory = getStep(items); + class CountingChunkListener implements ChunkListener { + int beforeCount = 0; + int afterCount = 0; + + public void afterChunk() { + afterCount++; + } + + public void beforeChunk() { + beforeCount++; + } + } + CountingChunkListener chunkListener = new CountingChunkListener(); + factory.setListeners(new BatchListener[]{ chunkListener }); + factory.setCommitInterval(commitInterval); + + ItemOrientedStep step = (ItemOrientedStep) factory.getObject(); + + job.setSteps(Collections.singletonList(step)); + + JobExecution jobExecution = repository.createJobExecution(job, new JobParameters()); + job.execute(jobExecution); + + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + assertNull(reader.read()); + assertEquals(items.length, written.size()); + + int expectedBeforeCount = (items.length / commitInterval ) + 1; + int expectedAdfterCount = (items.length / commitInterval); + assertEquals(expectedAdfterCount , chunkListener.afterCount); + assertEquals(expectedBeforeCount, chunkListener.beforeCount); + } }