From b215feb9e8067bb63cec6f4c599fcd503b2622fc Mon Sep 17 00:00:00 2001 From: dsyer Date: Mon, 2 Mar 2009 17:44:17 +0000 Subject: [PATCH] RESOLVED - issue BATCH-1111: ChunkListener called before WriteListener --- .../batch/core/step/AbstractStep.java | 9 ++-- .../step/item/BatchListenerFactoryHelper.java | 48 ------------------- .../core/step/item/SimpleStepFactoryBean.java | 10 ++-- .../batch/core/step/tasklet/TaskletStep.java | 27 ++++++++--- .../step/item/SimpleStepFactoryBeanTests.java | 31 +++++++++++- 5 files changed, 57 insertions(+), 68 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java index f51b7134d..dbe795e51 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java @@ -57,7 +57,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw private boolean allowStartIfComplete = false; - private CompositeStepExecutionListener listener = new CompositeStepExecutionListener(); + private CompositeStepExecutionListener stepExecutionListener = new CompositeStepExecutionListener(); private JobRepository jobRepository; @@ -192,7 +192,8 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw try { doExecute(stepExecution); - } catch (RepeatException e) { + } + catch (RepeatException e) { throw e.getCause(); } exitStatus = stepExecution.getExitStatus(); @@ -281,7 +282,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw * @param listener a {@link StepExecutionListener} */ public void registerStepExecutionListener(StepExecutionListener listener) { - this.listener.register(listener); + this.stepExecutionListener.register(listener); } /** @@ -299,7 +300,7 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw * @return composite listener that delegates to all registered listeners. */ protected StepExecutionListener getCompositeListener() { - return listener; + return stepExecutionListener; } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java index f1d24e845..e336c78fd 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/BatchListenerFactoryHelper.java @@ -18,14 +18,7 @@ package org.springframework.batch.core.step.item; import java.util.ArrayList; import java.util.List; -import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.StepListener; -import org.springframework.batch.core.listener.CompositeChunkListener; -import org.springframework.batch.repeat.RepeatContext; -import org.springframework.batch.repeat.RepeatOperations; -import org.springframework.batch.repeat.listener.RepeatListenerSupport; -import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.util.Assert; /** * Package private helper for step factory beans. @@ -35,47 +28,6 @@ import org.springframework.util.Assert; */ abstract class BatchListenerFactoryHelper { - /** - * @param chunkOperations - * @param listeners - */ - public static RepeatOperations addChunkListeners(RepeatOperations chunkOperations, StepListener[] listeners) { - - final CompositeChunkListener multicaster = new CompositeChunkListener(); - - boolean hasChunkListener = false; - - for (int i = 0; i < listeners.length; i++) { - StepListener listener = listeners[i]; - if (listener instanceof ChunkListener) { - hasChunkListener = true; - multicaster.register((ChunkListener) listener); - } - } - - if (hasChunkListener) { - - Assert.state(chunkOperations instanceof RepeatTemplate, - "Chunk operations is injected but not a RepeatTemplate, so chunk listeners cannot also be registered. " - + "Either inject a RepeatTemplate, or remove the ChunkListener."); - - RepeatTemplate stepTemplate = (RepeatTemplate) chunkOperations; - stepTemplate.registerListener(new RepeatListenerSupport() { - public void open(RepeatContext context) { - multicaster.beforeChunk(); - } - - public void close(RepeatContext context) { - multicaster.afterChunk(); - } - }); - - } - - return chunkOperations; - - } - public static List getListeners(StepListener[] listeners, Class cls) { List list = new ArrayList(); for (int i = 0; i < listeners.length; i++) { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java index 1b83bdf40..1ed7524e5 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java @@ -15,10 +15,6 @@ */ package org.springframework.batch.core.step.item; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.ChunkListener; @@ -524,7 +520,6 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { */ private void registerStepListeners(TaskletStep step, RepeatOperations chunkOperations) { - List chunkListeners = new ArrayList(Arrays.asList(getListeners())); for (Object itemHandler : new Object[] { itemReader, itemWriter, itemProcessor }) { if (itemHandler instanceof ItemStream) { step.registerStream((ItemStream) itemHandler); @@ -535,14 +530,15 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { step.registerStepExecutionListener((StepExecutionListener) listener); } if (listener instanceof ChunkListener) { - chunkListeners.add((StepListener) listener); + step.registerChunkListener((ChunkListener) listener); } } } - BatchListenerFactoryHelper.addChunkListeners(chunkOperations, chunkListeners.toArray(new StepListener[] {})); step.setStepExecutionListeners(BatchListenerFactoryHelper.getListeners(listeners, StepExecutionListener.class) .toArray(new StepExecutionListener[] {})); + step.setChunkListeners(BatchListenerFactoryHelper.getListeners(listeners, ChunkListener.class).toArray( + new ChunkListener[] {})); } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java index b1a5cf51a..6083be1df 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java @@ -20,10 +20,12 @@ import java.util.concurrent.Semaphore; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.core.listener.CompositeChunkListener; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.scope.context.ChunkContext; import org.springframework.batch.core.scope.context.StepContextRepeatCallback; @@ -69,6 +71,8 @@ public class TaskletStep extends AbstractStep { private RepeatOperations stepOperations = new RepeatTemplate(); + private CompositeChunkListener chunkListener = new CompositeChunkListener(); + // default to checking current thread for interruption. private StepInterruptionPolicy interruptionPolicy = new ThreadStepInterruptionPolicy(); @@ -143,17 +147,23 @@ public class TaskletStep extends AbstractStep { } /** - * Register each of the objects as listeners. If the {@link ItemReader} or - * {@link ItemWriter} themselves implements this interface they will be - * registered automatically, but their injected dependencies will not be. - * This is a good way to get access to job parameters and execution context - * if the tasklet is parameterised. + * Register a chunk listener for callbacks at the appropriate stages in a + * step execution. + * + * @param listener a {@link ChunkListener} + */ + public void registerChunkListener(ChunkListener listener) { + this.chunkListener.register(listener); + } + + /** + * Register each of the objects as listeners. * * @param listeners an array of listener objects of known types. */ - public void setStepExecutionListeners(StepExecutionListener[] listeners) { + public void setChunkListeners(ChunkListener[] listeners) { for (int i = 0; i < listeners.length; i++) { - registerStepExecutionListener(listeners[i]); + registerChunkListener(listeners[i]); } } @@ -242,6 +252,8 @@ public class TaskletStep extends AbstractStep { RepeatStatus result = RepeatStatus.CONTINUABLE; TransactionStatus transaction = transactionManager.getTransaction(transactionAttribute); + + chunkListener.beforeChunk(); boolean locked = false; @@ -249,6 +261,7 @@ public class TaskletStep extends AbstractStep { try { result = tasklet.execute(contribution, chunkContext); + chunkListener.afterChunk(); } finally { // Apply the contribution to the step diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java index af38e4360..331a34d52 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java @@ -39,6 +39,7 @@ import org.springframework.batch.core.Step; import org.springframework.batch.core.StepListener; import org.springframework.batch.core.job.SimpleJob; import org.springframework.batch.core.listener.ItemListenerSupport; +import org.springframework.batch.core.listener.StepListenerSupport; import org.springframework.batch.core.repository.dao.MapExecutionContextDao; import org.springframework.batch.core.repository.dao.MapJobExecutionDao; import org.springframework.batch.core.repository.dao.MapJobInstanceDao; @@ -217,21 +218,46 @@ public class SimpleStepFactoryBeanTests { int commitInterval = 3; SimpleStepFactoryBean factory = getStepFactory(items); + class AssertingWriteListener extends StepListenerSupport { + + String trail = ""; + + @Override + public void beforeWrite(List items) { + trail = trail + "2"; + } + + @Override + public void afterWrite(List items) { + trail = trail + "3"; + } + + } class CountingChunkListener implements ChunkListener { int beforeCount = 0; int afterCount = 0; + private AssertingWriteListener writeListener; + + public CountingChunkListener(AssertingWriteListener writeListener) { + super(); + this.writeListener = writeListener; + } + public void afterChunk() { + writeListener.trail = writeListener.trail + "4"; afterCount++; } public void beforeChunk() { + writeListener.trail = writeListener.trail + "1"; beforeCount++; } } - CountingChunkListener chunkListener = new CountingChunkListener(); - factory.setListeners(new StepListener[] { chunkListener }); + AssertingWriteListener writeListener = new AssertingWriteListener(); + CountingChunkListener chunkListener = new CountingChunkListener(writeListener); + factory.setListeners(new StepListener[] { chunkListener, writeListener }); factory.setCommitInterval(commitInterval); AbstractStep step = (AbstractStep) factory.getObject(); @@ -248,6 +274,7 @@ public class SimpleStepFactoryBeanTests { int expectedListenerCallCount = (items.length / commitInterval) + 1; assertEquals(expectedListenerCallCount, chunkListener.afterCount); assertEquals(expectedListenerCallCount, chunkListener.beforeCount); + assertTrue("Llistener order not as expected: " + writeListener.trail, writeListener.trail.startsWith("1234")); } /**