Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

IN PROGRESS - BATCH-1010: StepFactoryBeans cleanup

Browse Source
This commit is contained in:
robokaso
2009-02-10 16:29:35 +00:00
parent a321a2c21f
commit a079cbae4a
2 changed files with 41 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java
View File

@@ -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<T, S> extends SimpleStepFactoryBean<T,
private RetryContextCache retryContextCache;
private boolean isReaderTransactionalQueue = false;
public void setIsReaderTransactionalQueue(boolean isReaderTransactionalQueue) {
this.isReaderTransactionalQueue = isReaderTransactionalQueue;
}
/**
* Setter for the retry policy. If this is specified the other retry
* properties are ignored (retryLimit, backOffPolicy,
@@ -205,33 +200,19 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
@Override
protected void applyConfiguration(TaskletStep step) {
super.applyConfiguration(step);
if (!(retryLimit > 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<T> chunkProvider = configureChunkProvider();
SimpleChunkProcessor<T, S> chunkProcessor = configureChunkProcessor();
registerExplicitItemListeners(chunkProvider, chunkProcessor);
registerImplicitItemListeners(chunkProvider, chunkProcessor);
ChunkOrientedTasklet<T> tasklet = new ChunkOrientedTasklet<T>(chunkProvider, chunkProcessor);
tasklet.setBuffering(!isReaderTransactionalQueue);
step.setTasklet(tasklet);
super.applyConfiguration(step);
}
/**
* @return {@link ChunkProcessor} configured for fault-tolerance.
*/
private FaultTolerantChunkProcessor<T, S> configureChunkProcessor() {
@Override
protected FaultTolerantChunkProcessor<T, S> configureChunkProcessor() {
SkipPolicy writeSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, skippableExceptionClasses,
fatalExceptionClasses);
@@ -246,7 +227,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
FaultTolerantChunkProcessor<T, S> chunkProcessor = new FaultTolerantChunkProcessor<T, S>(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<T, S> extends SimpleStepFactoryBean<T,
/**
* @return {@link ChunkProvider} configured for fault-tolerance.
*/
private FaultTolerantChunkProvider<T> configureChunkProvider() {
@Override
protected FaultTolerantChunkProvider<T> configureChunkProvider() {
SkipPolicy readSkipPolicy = new LimitCheckingItemSkipPolicy(skipLimit, skippableExceptionClasses,
fatalExceptionClasses);

43
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
View File

@@ -112,6 +112,8 @@ public class SimpleStepFactoryBean<T, S> 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<T, S> 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<T, S> implements FactoryBean, BeanNameAware {
step.setStepOperations(stepOperations);
SimpleChunkProvider<T> chunkProvider = new SimpleChunkProvider<T>(itemReader, chunkOperations);
SimpleChunkProvider<T> chunkProvider = configureChunkProvider();
SimpleChunkProcessor<T, S> chunkProcessor = configureChunkProcessor();
SimpleChunkProcessor<T, S> chunkProcessor = new SimpleChunkProcessor<T, S>(itemProcessor, itemWriter);
registerExplicitItemListeners(chunkProvider, chunkProcessor);
registerImplicitItemListeners(chunkProvider, chunkProcessor);
ChunkOrientedTasklet<T> tasklet = new ChunkOrientedTasklet<T>(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<T, S> implements FactoryBean, BeanNameAware {
}
protected SimpleChunkProvider<T> configureChunkProvider() {
return new SimpleChunkProvider<T>(itemReader, chunkOperations);
}
protected SimpleChunkProcessor<T, S> configureChunkProcessor() {
return new SimpleChunkProcessor<T, S>(itemProcessor, itemWriter);
}
/**
* @return a {@link CompletionPolicy} consistent with the commit interval
* and injected policy (if present).
@@ -499,13 +520,14 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
* Register explicitly set ({@link #setListeners(StepListener[])}) item
* listeners.
*/
protected void registerExplicitItemListeners(SimpleChunkProvider<T> chunkProvider, SimpleChunkProcessor<T, S> chunkProcessor) {
protected void registerExplicitItemListeners(SimpleChunkProvider<T> chunkProvider,
SimpleChunkProcessor<T, S> chunkProcessor) {
chunkProvider.setListeners(BatchListenerFactoryHelper.<ItemReadListener<T>> getListeners(getListeners(),
ItemReadListener.class));
chunkProvider.setListeners(BatchListenerFactoryHelper.<SkipListener<T, S>> getListeners(getListeners(),
SkipListener.class));
chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemProcessListener<T, S>> getListeners(getListeners(),
ItemProcessListener.class));
chunkProcessor.setListeners(BatchListenerFactoryHelper.<ItemWriteListener<S>> getListeners(getListeners(),
@@ -518,9 +540,10 @@ public class SimpleStepFactoryBean<T, S> implements FactoryBean, BeanNameAware {
* Auto-register reader, processor and writer as item listeners if
* applicable.
*/
protected void registerImplicitItemListeners(SimpleChunkProvider<T> chunkProvider, SimpleChunkProcessor<T, S> chunkProcessor) {
protected void registerImplicitItemListeners(SimpleChunkProvider<T> chunkProvider,
SimpleChunkProcessor<T, S> chunkProcessor) {
for (Object itemHandler : new Object[] { getItemReader(), getItemWriter(), getItemProcessor() }) {
if (itemHandler instanceof SkipListener) {
chunkProvider.registerListener((StepListener) itemHandler);
chunkProcessor.registerListener((StepListener) itemHandler);