diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java index a946d142b..b605d5519 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/AbstractTaskletStepBuilder.java @@ -20,6 +20,7 @@ import java.util.Set; import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.core.step.tasklet.TaskletStep; import org.springframework.batch.item.ItemStream; @@ -45,7 +46,7 @@ import org.springframework.transaction.interceptor.TransactionAttribute; public abstract class AbstractTaskletStepBuilder> extends StepBuilderHelper> { - private Set listeners = new LinkedHashSet(); + protected Set chunkListeners = new LinkedHashSet(); private RepeatOperations stepOperations; @@ -72,12 +73,14 @@ public abstract class AbstractTaskletStepBuilder listener(ChunkListener listener) { - listeners.add(listener); + chunkListeners.add(listener); return this; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java index a7f01c60b..423079821 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/FaultTolerantStepBuilder.java @@ -27,6 +27,7 @@ import java.util.Set; import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.SkipListener; +import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.StepListener; import org.springframework.batch.core.listener.StepListenerFactoryBean; import org.springframework.batch.core.scope.context.ChunkContext; @@ -49,6 +50,7 @@ import org.springframework.batch.core.step.skip.SkipListenerFailedException; import org.springframework.batch.core.step.skip.SkipPolicy; import org.springframework.batch.core.step.skip.SkipPolicyFailedException; import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.core.step.tasklet.TaskletStep; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.repeat.RepeatOperations; @@ -133,6 +135,26 @@ public class FaultTolerantStepBuilder extends SimpleStepBuilder { super(parent); } + @Override + public TaskletStep build() { + registerStepListenerAsSkipListener(); + return super.build(); + } + + @SuppressWarnings("unchecked") + private void registerStepListenerAsSkipListener() { + for (StepExecutionListener stepExecutionListener: properties.getStepExecutionListeners()){ + if (stepExecutionListener instanceof SkipListener){ + listener((SkipListener)stepExecutionListener); + } + } + for (ChunkListener chunkListener: this.chunkListeners){ + if (chunkListener instanceof SkipListener){ + listener((SkipListener)chunkListener); + } + } + } + /** * Create a new chunk oriented tasklet with reader, writer and processor as provided. * diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java index 2b49a7f4b..697f86612 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/SimpleStepBuilder.java @@ -110,10 +110,33 @@ public class SimpleStepBuilder extends AbstractTaskletStepBuilder)stepListener); + } + if (stepListener instanceof ItemProcessListener){ + listener((ItemProcessListener)stepListener); + } + if (stepListener instanceof ItemWriteListener){ + listener((ItemWriteListener)stepListener); + } + } + @Override protected Tasklet createTasklet() { Assert.state(reader != null, "ItemReader must be provided"); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java index 03f2e16ec..157b2bc4f 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/builder/StepBuilderHelper.java @@ -39,7 +39,7 @@ public abstract class StepBuilderHelper> { protected final Log logger = LogFactory.getLog(getClass()); - private final CommonStepProperties properties; + protected final CommonStepProperties properties; public StepBuilderHelper(String name) { this.properties = new CommonStepProperties(); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/RegisterMultiListenerTest.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/RegisterMultiListenerTest.java new file mode 100644 index 000000000..79278a652 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/builder/RegisterMultiListenerTest.java @@ -0,0 +1,222 @@ +package org.springframework.batch.core.step.builder; + +import java.util.List; + +import javax.sql.DataSource; + +import static org.junit.Assert.*; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.ChunkListener; +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.ItemWriteListener; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.JobParametersInvalidException; +import org.springframework.batch.core.SkipListener; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; +import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; +import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException; +import org.springframework.batch.core.repository.JobRestartException; +import org.springframework.batch.core.scope.context.ChunkContext; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.NonTransientResourceException; +import org.springframework.batch.item.ParseException; +import org.springframework.batch.item.UnexpectedInputException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * Test for registering a listener class that implements different listeners interfaces + * just once in java based configuration. + * + * @author Tobias Flohre + */ +@ContextConfiguration(classes=RegisterMultiListenerTest.MultiListenerTestConfiguration.class) +@RunWith(SpringJUnit4ClassRunner.class) +public class RegisterMultiListenerTest { + + @Autowired + private JobLauncher jobLauncher; + + @Autowired + private Job job; + + @Autowired + private CallChecker callChecker; + + @Test + public void testMultiListener() throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException, JobParametersInvalidException{ + jobLauncher.run(job, new JobParameters()); + assertTrue("beforeStep hasn't been called",callChecker.beforeStepCalled); + assertTrue("beforeChunk hasn't been called",callChecker.beforeChunkCalled); + assertTrue("beforeWrite hasn't been called",callChecker.beforeWriteCalled); + assertTrue("skipInWrite hasn't been called",callChecker.skipInWriteCalled); + } + + @Configuration + @EnableBatchProcessing + public static class MultiListenerTestConfiguration{ + + @Autowired + private JobBuilderFactory jobBuilders; + + @Autowired + private StepBuilderFactory stepBuilders; + + @Bean + public Job testJob(){ + return jobBuilders.get("testJob") + .start(step()) + .build(); + } + + @Bean + public Step step(){ + return stepBuilders.get("step") + .listener(listener()) + .chunk(1) + .reader(reader()) + .writer(writer()) + .faultTolerant() + .skipLimit(1) + .skip(MySkippableException.class) + .build(); + } + + + @Bean + public DataSource dataSource(){ + EmbeddedDatabaseBuilder embeddedDatabaseBuilder = new EmbeddedDatabaseBuilder(); + return embeddedDatabaseBuilder.addScript("classpath:org/springframework/batch/core/schema-hsqldb.sql") + .setType(EmbeddedDatabaseType.HSQL) + .build(); + } + + @Bean + public CallChecker callChecker(){ + return new CallChecker(); + } + + @Bean + public MultiListener listener(){ + return new MultiListener(callChecker()); + } + + @Bean + public ItemReader reader(){ + return new ItemReader(){ + + @Override + public String read() throws Exception, + UnexpectedInputException, ParseException, + NonTransientResourceException { + return "item"; + } + + }; + } + + @Bean + public ItemWriter writer(){ + return new ItemWriter(){ + + @Override + public void write(List items) + throws Exception { + throw new MySkippableException(); + } + + }; + } + + } + + private static class CallChecker { + boolean beforeStepCalled = false; + boolean beforeChunkCalled = false; + boolean beforeWriteCalled = false; + boolean skipInWriteCalled = false; + } + + private static class MultiListener implements StepExecutionListener, ChunkListener, ItemWriteListener, SkipListener{ + + private CallChecker callChecker; + + private MultiListener(CallChecker callChecker) { + super(); + this.callChecker = callChecker; + } + + @Override + public void onSkipInRead(Throwable t) { + } + + @Override + public void onSkipInWrite(String item, Throwable t) { + callChecker.skipInWriteCalled = true; + } + + @Override + public void onSkipInProcess(String item, Throwable t) { + } + + @Override + public void beforeWrite(List items) { + callChecker.beforeWriteCalled = true; + } + + @Override + public void afterWrite(List items) { + } + + @Override + public void onWriteError(Exception exception, + List items) { + } + + @Override + public void beforeChunk(ChunkContext context) { + callChecker.beforeChunkCalled = true; + } + + @Override + public void afterChunk(ChunkContext context) { + } + + @Override + public void afterChunkError(ChunkContext context) { + } + + @Override + public void beforeStep(StepExecution stepExecution) { + callChecker.beforeStepCalled = true; + } + + @Override + public ExitStatus afterStep(StepExecution stepExecution) { + return null; + } + + } + + private static class MySkippableException extends RuntimeException{ + + private static final long serialVersionUID = 1L; + + } + +}