From 1412135c0b286a0740ac4857ece8730228950f4e Mon Sep 17 00:00:00 2001 From: robokaso Date: Fri, 31 Oct 2008 09:27:47 +0000 Subject: [PATCH] IN PROGRESS - BATCH-896: "DRY" FaultTolerantTasklet implementations --- ...ractFaultTolerantChunkOrientedTasklet.java | 95 +++++++++++++++++++ .../item/AbstractItemOrientedTasklet.java | 79 --------------- .../FaultTolerantChunkOrientedTasklet.java | 20 ++-- ...ringFaultTolerantChunkOrientedTasklet.java | 2 +- 4 files changed, 106 insertions(+), 90 deletions(-) create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java new file mode 100644 index 000000000..2f762b9ea --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractFaultTolerantChunkOrientedTasklet.java @@ -0,0 +1,95 @@ +package org.springframework.batch.core.step.item; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; + +import org.springframework.batch.core.step.skip.SkipListenerFailedException; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.core.AttributeAccessor; + +public abstract class AbstractFaultTolerantChunkOrientedTasklet extends AbstractItemOrientedTasklet { + + public AbstractFaultTolerantChunkOrientedTasklet(ItemReader itemReader, + ItemProcessor itemProcessor, ItemWriter itemWriter) { + super(itemReader, itemProcessor, itemWriter); + } + + /** + * Call all skip listeners in read-process-write order + * @param skippedReads read exceptions + * @param skippedInputs items and corresponding exceptions skipped in + * processing phase + * @param skippedOutputs items and corresponding exceptions skipped in write + * phase + */ + protected void callSkipListeners(final List skippedReads, final Map skippedInputs, + final Map skippedOutputs) { + + for (Exception e : skippedReads) { + try { + listener.onSkipInRead(e); + } + catch (RuntimeException ex) { + throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e); + } + } + for (Entry skip : skippedInputs.entrySet()) { + try { + listener.onSkipInProcess(skip.getKey(), skip.getValue()); + } + catch (RuntimeException ex) { + throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, skip.getValue()); + } + } + + for (Entry skip : skippedOutputs.entrySet()) { + try { + listener.onSkipInWrite(skip.getKey(), skip.getValue()); + } + catch (RuntimeException ex) { + throw new SkipListenerFailedException("Fatal exception in skip listener", ex, skip.getValue()); + } + } + } + + /** + * Return a list stored in the attributes under the key. Create an empty + * list and store it if the list is not stored yet. + */ + protected static List getBufferedList(AttributeAccessor attributes, String key) { + List buffer; + if (!attributes.hasAttribute(key)) { + buffer = new ArrayList(); + attributes.setAttribute(key, buffer); + } + else { + @SuppressWarnings("unchecked") + List casted = (List) attributes.getAttribute(key); + buffer = casted; + } + return buffer; + } + + /** + * Return a map of items to exceptions stored in the attributes under the + * key, Create an empty map and store it if the list is not stored yet. + */ + protected static Map getBufferedSkips(AttributeAccessor attributes, String key) { + Map buffer; + if (!attributes.hasAttribute(key)) { + buffer = new LinkedHashMap(); + attributes.setAttribute(key, buffer); + } + else { + @SuppressWarnings("unchecked") + Map casted = (Map) attributes.getAttribute(key); + buffer = casted; + } + return buffer; + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemOrientedTasklet.java index 1282b5f4b..d08480ad6 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemOrientedTasklet.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemOrientedTasklet.java @@ -1,21 +1,15 @@ package org.springframework.batch.core.step.item; -import java.util.ArrayList; -import java.util.LinkedHashMap; import java.util.List; -import java.util.Map; -import java.util.Map.Entry; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.StepListener; import org.springframework.batch.core.listener.MulticasterBatchListener; -import org.springframework.batch.core.step.skip.SkipListenerFailedException; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; -import org.springframework.core.AttributeAccessor; /** * Superclass for {@link Tasklet}s implementing variations on read-process-write @@ -120,77 +114,4 @@ public abstract class AbstractItemOrientedTasklet implements Tasklet { } } - /** - * Call all skip listeners in read-process-write order - * @param skippedReads read exceptions - * @param skippedInputs items and corresponding exceptions skipped in - * processing phase - * @param skippedOutputs items and corresponding exceptions skipped in write - * phase - */ - protected void callSkipListeners(final List skippedReads, final Map skippedInputs, - final Map skippedOutputs) { - - for (Exception e : skippedReads) { - try { - listener.onSkipInRead(e); - } - catch (RuntimeException ex) { - throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e); - } - } - for (Entry skip : skippedInputs.entrySet()) { - try { - listener.onSkipInProcess(skip.getKey(), skip.getValue()); - } - catch (RuntimeException ex) { - throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, skip.getValue()); - } - } - - for (Entry skip : skippedOutputs.entrySet()) { - try { - listener.onSkipInWrite(skip.getKey(), skip.getValue()); - } - catch (RuntimeException ex) { - throw new SkipListenerFailedException("Fatal exception in skip listener", ex, skip.getValue()); - } - } - } - - /** - * Return a list stored in the attributes under the key. Create an empty - * list and store it if the list is not stored yet. - */ - protected static List getBufferedList(AttributeAccessor attributes, String key) { - List buffer; - if (!attributes.hasAttribute(key)) { - buffer = new ArrayList(); - attributes.setAttribute(key, buffer); - } - else { - @SuppressWarnings("unchecked") - List casted = (List) attributes.getAttribute(key); - buffer = casted; - } - return buffer; - } - - /** - * Return a map of items to exceptions stored in the attributes under the - * key, Create an empty map and store it if the list is not stored yet. - */ - protected static Map getBufferedSkips(AttributeAccessor attributes, String key) { - Map buffer; - if (!attributes.hasAttribute(key)) { - buffer = new LinkedHashMap(); - attributes.setAttribute(key, buffer); - } - else { - @SuppressWarnings("unchecked") - Map casted = (Map) attributes.getAttribute(key); - buffer = casted; - } - return buffer; - } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java index f74a4fca1..42fb5b83d 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java @@ -55,7 +55,7 @@ import org.springframework.core.AttributeAccessor; * @author Dave Syer * @author Robert Kasanicky */ -public class FaultTolerantChunkOrientedTasklet extends AbstractItemOrientedTasklet { +public class FaultTolerantChunkOrientedTasklet extends AbstractFaultTolerantChunkOrientedTasklet { private static final String INPUT_BUFFER_KEY = "INPUT_BUFFER_KEY"; @@ -77,8 +77,8 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente private static final String SKIPPED_READS_KEY = "SKIPPED_READS_BUFFER_KEY"; - public FaultTolerantChunkOrientedTasklet(ItemReader itemReader, - ItemProcessor itemProcessor, ItemWriter itemWriter, + public FaultTolerantChunkOrientedTasklet(ItemReader itemReader, + ItemProcessor itemProcessor, ItemWriter itemWriter, RepeatOperations chunkOperations, RetryOperations retryTemplate, Classifier rollbackClassifier, ItemSkipPolicy readSkipPolicy, ItemSkipPolicy writeSkipPolicy, ItemSkipPolicy processSkipPolicy) { @@ -102,7 +102,7 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente */ public ExitStatus execute(final StepContribution contribution, AttributeAccessor attributes) throws Exception { - final List inputs = getBufferedList(attributes, INPUT_BUFFER_KEY); + final List inputs = getBufferedList(attributes, INPUT_BUFFER_KEY); final List outputs = new ArrayList(); ExitStatus result = ExitStatus.CONTINUABLE; @@ -113,7 +113,7 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente result = repeatOperations.iterate(new RepeatCallback() { public ExitStatus doInIteration(final RepeatContext context) throws Exception { - T item = read(contribution, skippedReads); + I item = read(contribution, skippedReads); if (item == null) { return ExitStatus.FINISHED; @@ -131,7 +131,7 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente } - Map skippedInputs = getBufferedSkips(attributes, SKIPPED_INPUTS_KEY); + Map skippedInputs = getBufferedSkips(attributes, SKIPPED_INPUTS_KEY); if (!inputs.isEmpty()) { inputs.removeAll(skippedInputs.keySet()); process(contribution, inputs, outputs, skippedInputs); @@ -167,7 +167,7 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente * @param skippedReads * @return next item for processing */ - protected T read(StepContribution contribution, List skippedReads) throws Exception { + protected I read(StepContribution contribution, List skippedReads) throws Exception { while (true) { try { @@ -200,12 +200,12 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente /** * Incorporate retry into the item processor stage. */ - protected void process(final StepContribution contribution, final List inputs, final List outputs, - final Map skippedInputs) throws Exception { + protected void process(final StepContribution contribution, final List inputs, final List outputs, + final Map skippedInputs) throws Exception { int filtered = 0; - for (final T item : inputs) { + for (final I item : inputs) { RetryCallback retryCallback = new RetryCallback() { diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java index 70262c5ec..71f051542 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java @@ -39,7 +39,7 @@ import org.springframework.core.AttributeAccessor; * @param input item type * @param output item type */ -public class NonbufferingFaultTolerantChunkOrientedTasklet extends AbstractItemOrientedTasklet { +public class NonbufferingFaultTolerantChunkOrientedTasklet extends AbstractFaultTolerantChunkOrientedTasklet { private static final String SKIPPED_INPUTS_KEY = "SKIPPED_INPUTS_KEY";