From a079cbae4af6ea6bea9897fade67ad28292e5e35 Mon Sep 17 00:00:00 2001 From: robokaso Date: Tue, 10 Feb 2009 16:29:35 +0000 Subject: [PATCH] IN PROGRESS - BATCH-1010: StepFactoryBeans cleanup --- .../item/FaultTolerantStepFactoryBean.java | 34 ++++----------- .../core/step/item/SimpleStepFactoryBean.java | 43 ++++++++++++++----- 2 files changed, 41 insertions(+), 36 deletions(-) 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 e99677d10..8f7d17dfb 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 @@ -21,6 +21,7 @@ import java.util.Collection; import java.util.HashSet; import java.util.List; +import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; import org.springframework.batch.core.step.skip.NonSkippableReadException; import org.springframework.batch.core.step.skip.SkipLimitExceededException; @@ -87,12 +88,6 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean extends SimpleStepFactoryBean 0 || skipLimit > 0 || retryPolicy != null)) { - // zero fault-tolerance, just use the parent's simple config - return; - } addFatalExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class, - SkipListenerFailedException.class, RetryException.class); + SkipListenerFailedException.class, RetryException.class, JobInterruptedException.class); - SimpleChunkProvider chunkProvider = configureChunkProvider(); - SimpleChunkProcessor chunkProcessor = configureChunkProcessor(); - - registerExplicitItemListeners(chunkProvider, chunkProcessor); - registerImplicitItemListeners(chunkProvider, chunkProcessor); - - ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); - tasklet.setBuffering(!isReaderTransactionalQueue); - - step.setTasklet(tasklet); + super.applyConfiguration(step); } /** * @return {@link ChunkProcessor} configured for fault-tolerance. */ - private FaultTolerantChunkProcessor configureChunkProcessor() { + @Override + protected FaultTolerantChunkProcessor configureChunkProcessor() { SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, skippableExceptionClasses, fatalExceptionClasses); @@ -246,7 +227,7 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean chunkProcessor = new FaultTolerantChunkProcessor(getItemProcessor(), getItemWriter(), batchRetryTemplate); - chunkProcessor.setBuffering(!isReaderTransactionalQueue); + chunkProcessor.setBuffering(!isReaderTransactionalQueue()); chunkProcessor.setWriteSkipPolicy(writeSkipPolicy); chunkProcessor.setProcessSkipPolicy(writeSkipPolicy); chunkProcessor.setRollbackClassifier(rollbackClassifier); @@ -257,7 +238,8 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean configureChunkProvider() { + @Override + protected FaultTolerantChunkProvider configureChunkProvider() { SkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, skippableExceptionClasses, fatalExceptionClasses); 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 b5775810d..be1f9300d 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 @@ -112,6 +112,8 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { private int throttleLimit = TaskExecutorRepeatTemplate.DEFAULT_THROTTLE_LIMIT; + private boolean isReaderTransactionalQueue = false; + /** * Default constructor for {@link SimpleStepFactoryBean}. */ @@ -119,9 +121,19 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { super(); } + public void setIsReaderTransactionalQueue(boolean isReaderTransactionalQueue) { + this.isReaderTransactionalQueue = isReaderTransactionalQueue; + } + + protected boolean isReaderTransactionalQueue() { + return isReaderTransactionalQueue; + } + /** - * Set the bean name property, which will become the name of the - * {@link Step} when it is created. + * public void setIsReaderTransactionalQueue(boolean + * isReaderTransactionalQueue) { this.isReaderTransactionalQueue = + * isReaderTransactionalQueue; } Set the bean name property, which will + * become the name of the {@link Step} when it is created. * * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String) */ @@ -443,14 +455,15 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { step.setStepOperations(stepOperations); - SimpleChunkProvider chunkProvider = new SimpleChunkProvider(itemReader, chunkOperations); + SimpleChunkProvider chunkProvider = configureChunkProvider(); + + SimpleChunkProcessor chunkProcessor = configureChunkProcessor(); - SimpleChunkProcessor chunkProcessor = new SimpleChunkProcessor(itemProcessor, itemWriter); - registerExplicitItemListeners(chunkProvider, chunkProcessor); registerImplicitItemListeners(chunkProvider, chunkProcessor); ChunkOrientedTasklet tasklet = new ChunkOrientedTasklet(chunkProvider, chunkProcessor); + tasklet.setBuffering(!isReaderTransactionalQueue()); // 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 @@ -476,6 +489,14 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { } + protected SimpleChunkProvider configureChunkProvider() { + return new SimpleChunkProvider(itemReader, chunkOperations); + } + + protected SimpleChunkProcessor configureChunkProcessor() { + return new SimpleChunkProcessor(itemProcessor, itemWriter); + } + /** * @return a {@link CompletionPolicy} consistent with the commit interval * and injected policy (if present). @@ -499,13 +520,14 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { * Register explicitly set ({@link #setListeners(StepListener[])}) item * listeners. */ - protected void registerExplicitItemListeners(SimpleChunkProvider chunkProvider, SimpleChunkProcessor chunkProcessor) { - + protected void registerExplicitItemListeners(SimpleChunkProvider chunkProvider, + SimpleChunkProcessor chunkProcessor) { + chunkProvider.setListeners(BatchListenerFactoryHelper.> getListeners(getListeners(), ItemReadListener.class)); chunkProvider.setListeners(BatchListenerFactoryHelper.> getListeners(getListeners(), SkipListener.class)); - + chunkProcessor.setListeners(BatchListenerFactoryHelper.> getListeners(getListeners(), ItemProcessListener.class)); chunkProcessor.setListeners(BatchListenerFactoryHelper.> getListeners(getListeners(), @@ -518,9 +540,10 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { * Auto-register reader, processor and writer as item listeners if * applicable. */ - protected void registerImplicitItemListeners(SimpleChunkProvider chunkProvider, SimpleChunkProcessor chunkProcessor) { + protected void registerImplicitItemListeners(SimpleChunkProvider chunkProvider, + SimpleChunkProcessor chunkProcessor) { for (Object itemHandler : new Object[] { getItemReader(), getItemWriter(), getItemProcessor() }) { - + if (itemHandler instanceof SkipListener) { chunkProvider.registerListener((StepListener) itemHandler); chunkProcessor.registerListener((StepListener) itemHandler);