diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java index 5802f8375..31d7b0192 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java @@ -25,6 +25,7 @@ import org.springframework.batch.core.ItemProcessListener; import org.springframework.batch.core.ItemReadListener; import org.springframework.batch.core.ItemWriteListener; import org.springframework.batch.core.SkipListener; +import org.springframework.batch.core.StepListener; import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; import org.springframework.batch.core.step.skip.NonSkippableReadException; import org.springframework.batch.core.step.skip.SkipLimitExceededException; @@ -312,6 +313,22 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean> getListeners(getListeners(), SkipListener.class)); + + for (Object itemHandler : new Object[] { getItemReader(), getItemWriter(), getItemProcessor() }) { + + if (itemHandler instanceof SkipListener) { + chunkProvider.registerListener((StepListener) itemHandler); + chunkProcessor.registerListener((StepListener) itemHandler); + // already registered with both so avoid double-registering + continue; + } + if (itemHandler instanceof ItemReadListener) { + chunkProvider.registerListener((StepListener) itemHandler); + } + if (itemHandler instanceof ItemProcessListener || itemHandler instanceof ItemWriteListener) { + chunkProcessor.registerListener((StepListener) itemHandler); + } + } ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); tasklet.setBuffering(!isReaderTransactionalQueue); 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 fa0df5091..1ba6de926 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,13 @@ */ 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; import org.springframework.batch.core.ItemProcessListener; import org.springframework.batch.core.ItemReadListener; import org.springframework.batch.core.ItemWriteListener; @@ -417,42 +420,11 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { ItemWriter itemWriter = getItemWriter(); ItemProcessor itemProcessor = getItemProcessor(); - // Since we are going to wrap these things with listener callbacks we - // need to register them here because the step will not know we did - // that. - if (itemReader instanceof ItemStream) { - step.registerStream((ItemStream) itemReader); - } - if (itemReader instanceof StepExecutionListener) { - step.registerStepExecutionListener((StepExecutionListener) itemReader); - } - if (itemProcessor instanceof ItemStream) { - step.registerStream((ItemStream) itemProcessor); - } - if (itemProcessor instanceof StepExecutionListener) { - step.registerStepExecutionListener((StepExecutionListener) itemProcessor); - } - if (itemWriter instanceof ItemStream) { - step.registerStream((ItemStream) itemWriter); - } - if (itemWriter instanceof StepExecutionListener) { - step.registerStepExecutionListener((StepExecutionListener) itemWriter); - } - - List array = BatchListenerFactoryHelper.getListeners(listeners, - StepExecutionListener.class); - StepExecutionListener[] stepListeners = new StepExecutionListener[array.size()]; - for (int i = 0; i < stepListeners.length; i++) { - stepListeners[i] = array.get(i); - } - step.setStepExecutionListeners(stepListeners); - if (chunkOperations == null) { RepeatTemplate repeatTemplate = new RepeatTemplate(); repeatTemplate.setCompletionPolicy(getChunkCompletionPolicy()); chunkOperations = repeatTemplate; } - BatchListenerFactoryHelper.addChunkListeners(chunkOperations, getListeners()); if (stepOperations == null) { @@ -471,15 +443,47 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { step.setStepOperations(stepOperations); + SimpleChunkProvider chunkProvider = new SimpleChunkProvider(itemReader, chunkOperations); + List> readListeners = BatchListenerFactoryHelper.>getListeners(getListeners(), ItemReadListener.class); + chunkProvider.setListeners(readListeners); + SimpleChunkProcessor chunkProcessor = new SimpleChunkProcessor(itemProcessor, itemWriter); chunkProcessor.setListeners(BatchListenerFactoryHelper.>getListeners(getListeners(), ItemProcessListener.class)); chunkProcessor.setListeners(BatchListenerFactoryHelper.>getListeners(getListeners(), ItemWriteListener.class)); - SimpleChunkProvider chunkProvider = new SimpleChunkProvider(itemReader, chunkOperations); - List> readListeners = BatchListenerFactoryHelper.>getListeners(getListeners(), ItemReadListener.class); - chunkProvider.setListeners(readListeners); ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); + // Since we are going to wrap these things with listener callbacks we + // need to register them here because the step will not know we did + // that. + List chunkListeners = new ArrayList(Arrays.asList(getListeners())); + for(Object itemHandler: new Object[]{itemReader, itemWriter, itemProcessor}){ + if (itemHandler instanceof ItemStream) { + step.registerStream((ItemStream) itemHandler); + } + if (itemHandler instanceof StepExecutionListener) { + step.registerStepExecutionListener((StepExecutionListener) itemHandler); + } + if (itemHandler instanceof ChunkListener) { + chunkListeners.add((StepListener) itemHandler); + } + if (itemHandler instanceof ItemReadListener) { + chunkProvider.registerListener((StepListener) itemHandler); + } + if (itemHandler instanceof ItemProcessListener || itemHandler instanceof ItemWriteListener) { + chunkProcessor.registerListener((StepListener) itemHandler); + } + } + + BatchListenerFactoryHelper.addChunkListeners(chunkOperations, chunkListeners.toArray(new StepListener[]{})); + List array = BatchListenerFactoryHelper.getListeners(listeners, + StepExecutionListener.class); + StepExecutionListener[] stepListeners = new StepExecutionListener[array.size()]; + for (int i = 0; i < stepListeners.length; i++) { + stepListeners[i] = array.get(i); + } + step.setStepExecutionListeners(stepListeners); + step.setTasklet(tasklet); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java index b250cba58..1be22784f 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBeanTests.java @@ -14,9 +14,14 @@ import org.apache.commons.logging.LogFactory; import org.junit.Before; import org.junit.Test; import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.ItemProcessListener; +import org.springframework.batch.core.ItemReadListener; +import org.springframework.batch.core.ItemWriteListener; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.SkipListener; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.StepListener; @@ -602,6 +607,81 @@ public class FaultTolerantStepFactoryBeanTests { } + @Test + public void testAutoRegisterItemListeners() throws Exception { + + final List listenerCalls = new ArrayList(); + + class TestItemListenerWriter implements ItemWriter, ItemReadListener, + ItemWriteListener, ItemProcessListener, SkipListener, + ChunkListener { + public void write(List items) throws Exception { + if (items.contains("4")) { + throw new SkippableException("skippable"); + } + } + + public void afterRead(String item) { + listenerCalls.add(1); + } + + public void beforeRead() { + } + + public void onReadError(Exception ex) { + } + + public void afterWrite(List items) { + listenerCalls.add(2); + } + + public void beforeWrite(List items) { + } + + public void onWriteError(Exception exception, List items) { + } + + public void afterProcess(String item, String result) { + listenerCalls.add(3); + } + + public void beforeProcess(String item) { + } + + public void onProcessError(String item, Exception e) { + } + + public void afterChunk() { + listenerCalls.add(4); + } + + public void beforeChunk() { + } + + public void onSkipInProcess(String item, Throwable t) { + } + + public void onSkipInRead(Throwable t) { + listenerCalls.add(6); + } + + public void onSkipInWrite(String item, Throwable t) { + listenerCalls.add(5); + } + + } + + factory.setItemWriter(new TestItemListenerWriter()); + + Step step = (Step) factory.getObject(); + step.execute(stepExecution); + + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + for (int i = 1; i <= 6; i++) { + assertTrue("didn't call listener " + i, listenerCalls.contains(i)); + } + } + private static class SkipProcessorStub implements ItemProcessor { private final Collection failures; 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 03acf995a..d48829df2 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 @@ -30,6 +30,9 @@ import org.junit.Before; import org.junit.Test; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.ChunkListener; +import org.springframework.batch.core.ItemProcessListener; +import org.springframework.batch.core.ItemReadListener; +import org.springframework.batch.core.ItemWriteListener; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.Step; @@ -81,7 +84,7 @@ public class SimpleStepFactoryBeanTests { MapStepExecutionDao.clear(); } - @Test(expected=IllegalArgumentException.class) + @Test(expected = IllegalArgumentException.class) public void testMandatoryProperties() throws Exception { new SimpleStepFactoryBean().getObject(); } @@ -291,6 +294,73 @@ public class SimpleStepFactoryBeanTests { } + @Test + public void testAutoRegisterItemListeners() throws Exception { + + SimpleStepFactoryBean factory = getStepFactory(new String[] { "foo", "bar", "spam" }); + + final List listenerCalls = new ArrayList(); + + class TestItemListenerWriter implements ItemWriter, ItemReadListener, + ItemWriteListener, ItemProcessListener, ChunkListener { + public void write(List items) throws Exception { + } + + public void afterRead(String item) { + listenerCalls.add(1); + } + + public void beforeRead() { + } + + public void onReadError(Exception ex) { + } + + public void afterWrite(List items) { + listenerCalls.add(2); + } + + public void beforeWrite(List items) { + } + + public void onWriteError(Exception exception, List items) { + } + + public void afterProcess(String item, String result) { + listenerCalls.add(3); + } + + public void beforeProcess(String item) { + } + + public void onProcessError(String item, Exception e) { + } + + public void afterChunk() { + listenerCalls.add(4); + } + + public void beforeChunk() { + } + + } + + factory.setItemWriter(new TestItemListenerWriter()); + + Step step = (Step) factory.getObject(); + + job.setSteps(Collections.singletonList(step)); + + JobExecution jobExecution = repository.createJobExecution(job.getName(), new JobParameters()); + + job.execute(jobExecution); + + assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus()); + for (int i = 1; i <= 4; i++) { + assertTrue(listenerCalls.contains(i)); + } + } + private SimpleStepFactoryBean getStepFactory(String... args) throws Exception { SimpleStepFactoryBean factory = new SimpleStepFactoryBean();