Merge pull request #211 from tobiasflohre/BATCH-2077

* BATCH-2077: BATCH-2077 Java Config: Register listeners just once
2013-08-09 14:44:49 -05:00
parent c252b132f8 4d9b75359d
commit 60709cd923
5 changed files with 282 additions and 4 deletions
--- 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<B extends AbstractTaskletStepBuilder<B>> extends
 		StepBuilderHelper<AbstractTaskletStepBuilder<B>> {

-	private Set<ChunkListener> listeners = new LinkedHashSet<ChunkListener>();
+	protected Set<ChunkListener> chunkListeners = new LinkedHashSet<ChunkListener>();

 	private RepeatOperations stepOperations;

@@ -72,12 +73,14 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
 	 * @return a tasklet step fully configured and read to execute
 	 */
 	public TaskletStep build() {
+		
+		registerStepListenerAsChunkListener();

 		TaskletStep step = new TaskletStep(getName());

 		super.enhance(step);

-		step.setChunkListeners(listeners.toArray(new ChunkListener[0]));
+		step.setChunkListeners(chunkListeners.toArray(new ChunkListener[0]));

 		if (transactionAttribute != null) {
 			step.setTransactionAttribute(transactionAttribute);
@@ -113,6 +116,14 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu

 	}

+	private void registerStepListenerAsChunkListener() {
+		for (StepExecutionListener stepExecutionListener: properties.getStepExecutionListeners()){
+			if (stepExecutionListener instanceof ChunkListener){
+				listener((ChunkListener)stepExecutionListener);
+			}
+		}
+	}
+
 	/**
 	 * Register a chunk listener.
 	 * 
@@ -120,7 +131,7 @@ public abstract class AbstractTaskletStepBuilder<B extends AbstractTaskletStepBu
 	 * @return this for fluent chaining
 	 */
 	public AbstractTaskletStepBuilder<B> listener(ChunkListener listener) {
-		listeners.add(listener);
+		chunkListeners.add(listener);
 		return this;
 	}

--- 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<I, O> extends SimpleStepBuilder<I, O> {
 		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<I,O>)stepExecutionListener);
+			}
+		}
+		for (ChunkListener chunkListener: this.chunkListeners){
+			if (chunkListener instanceof SkipListener){
+				listener((SkipListener<I,O>)chunkListener);
+			}
+		}
+	}
+
 	/**
 	 * Create a new chunk oriented tasklet with reader, writer and processor as provided.
 	 *
--- 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<I, O> extends AbstractTaskletStepBuilder<SimpleSt
 	 */
 	@Override
 	public TaskletStep build() {
+		registerStepListenerAsItemListener();
 		registerAsStreamsAndListeners(reader, processor, writer);
 		return super.build();
 	}

+	private void registerStepListenerAsItemListener() {
+		for (StepExecutionListener stepExecutionListener: properties.getStepExecutionListeners()){
+			checkAndAddItemListener(stepExecutionListener);
+		}
+		for (ChunkListener chunkListener: this.chunkListeners){
+			checkAndAddItemListener(chunkListener);
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private void checkAndAddItemListener(StepListener stepListener) {
+		if (stepListener instanceof ItemReadListener){
+			listener((ItemReadListener<I>)stepListener);
+		}
+		if (stepListener instanceof ItemProcessListener){
+			listener((ItemProcessListener<I,O>)stepListener);
+		}
+		if (stepListener instanceof ItemWriteListener){
+			listener((ItemWriteListener<O>)stepListener);
+		}
+	}
+
 	@Override
 	protected Tasklet createTasklet() {
 		Assert.state(reader != null, "ItemReader must be provided");
--- 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<B extends StepBuilderHelper<B>> {

 	protected final Log logger = LogFactory.getLog(getClass());

-	private final CommonStepProperties properties;
+	protected final CommonStepProperties properties;

 	public StepBuilderHelper(String name) {
 		this.properties = new CommonStepProperties();
--- 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
@@ -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())
+					.<String,String>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<String> reader(){
+			return new ItemReader<String>(){
+
+				@Override
+				public String read() throws Exception,
+						UnexpectedInputException, ParseException,
+						NonTransientResourceException {
+					return "item";
+				}
+				
+			};
+		}
+		
+		@Bean
+		public ItemWriter<String> writer(){
+			return new ItemWriter<String>(){
+
+				@Override
+				public void write(List<? extends String> 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<String>, SkipListener<String,String>{
+		
+		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<? extends String> items) {
+			callChecker.beforeWriteCalled = true;
+		}
+
+		@Override
+		public void afterWrite(List<? extends String> items) {
+		}
+
+		@Override
+		public void onWriteError(Exception exception,
+				List<? extends String> 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;
+		
+	}
+
+}