From c4291f10af1a73d840dcab852ba129c6dff278bf Mon Sep 17 00:00:00 2001 From: robokaso Date: Wed, 17 Sep 2008 11:39:29 +0000 Subject: [PATCH] OPEN - BATCH-803: Add non-buffering ChunkOrientedTasklet (or option in existing one) plus flag for factory bean comments cleanup --- .../FaultTolerantChunkOrientedTasklet.java | 54 +++++++++---------- 1 file changed, 25 insertions(+), 29 deletions(-) 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 819364878..4613f94e4 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 @@ -39,26 +39,25 @@ import org.springframework.batch.support.Classifier; import org.springframework.core.AttributeAccessor; /** - * If there is an exception on input it is skipped if allowed. If there is - * an exception on output, it will be re-thrown in any case, and the - * behaviour when the item is next encountered depends on the retryable and - * skippable exception configuration. If the exception is retryable the - * write will be attempted again up to the retry limit. When retry attempts - * are exhausted the skip listener is invoked and the skip count - * incremented. A retryable exception is thus also effectively also - * implicitly skippable. + * If there is an exception on input it is skipped if allowed. If there is an + * exception on output, it will be re-thrown in any case, and the behaviour when + * the item is next encountered depends on the retryable and skippable exception + * configuration. If the exception is retryable the write will be attempted + * again up to the retry limit. When retry attempts are exhausted the skip + * listener is invoked and the skip count incremented. A retryable exception is + * thus also effectively also implicitly skippable. * * @author Dave Syer * @author Robert Kasanicky */ -public class FaultTolerantChunkOrientedTasklet extends AbstractItemOrientedTasklet { +public class FaultTolerantChunkOrientedTasklet extends AbstractItemOrientedTasklet { private static final String INPUT_BUFFER_KEY = "INPUT_BUFFER_KEY"; private static final String OUTPUT_BUFFER_KEY = "OUTPUT_BUFFER_KEY"; private final RepeatOperations repeatOperations; - + final private RetryOperations retryOperations; final private ItemSkipPolicy readSkipPolicy; @@ -69,7 +68,6 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente final private Classifier rollbackClassifier; - public FaultTolerantChunkOrientedTasklet(ItemReader itemReader, ItemProcessor itemProcessor, ItemWriter itemWriter, RepeatOperations chunkOperations, RetryOperations retryTemplate, @@ -95,7 +93,8 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente */ public ExitStatus execute(final StepContribution contribution, AttributeAccessor attributes) throws Exception { - // TODO: check flags to see if these need to be saved or not (e.g. JMS not) + // TODO: check flags to see if these need to be saved or not (e.g. JMS + // not) final Chunk inputs = getInputBuffer(attributes); final Chunk outputs = getOutputBuffer(attributes); @@ -106,7 +105,7 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente result = repeatOperations.iterate(new RepeatCallback() { public ExitStatus doInIteration(final RepeatContext context) throws Exception { T item = read(contribution); - + if (item == null) { return ExitStatus.FINISHED; } @@ -115,16 +114,16 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente return ExitStatus.CONTINUABLE; } }); - + // If there is no input we don't have to do anything more if (inputs.isEmpty()) { return result; } - + storeInputs(attributes, inputs); } - + if (!inputs.isEmpty()) { process(contribution, inputs, outputs); } @@ -149,8 +148,7 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente * Tries to read the item from the reader, in case of exception skip the * item if the skip policy allows, otherwise re-throw. * - * @param contribution current StepContribution holding skipped items - * count + * @param contribution current StepContribution holding skipped items count * @return next item for processing */ protected T read(StepContribution contribution) throws Exception { @@ -173,14 +171,12 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente logger.debug("Skipping failed input", e); } else { - // re-throw only when the skip policy runs out of - // patience + // skip doesn't apply -> rethrow throw e; } } catch (SkipLimitExceededException ex) { - // we are headed for a abnormal ending so bake in the - // skip count + // re-throw when the skip policy runs out of patience throw ex; } } @@ -235,7 +231,8 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente }; - S output = retryOperations.execute(retryCallback, recoveryCallback, new DefaultRetryState(item, rollbackClassifier)); + S output = retryOperations.execute(retryCallback, recoveryCallback, new DefaultRetryState(item, + rollbackClassifier)); if (output != null) { outputs.add(output); } @@ -264,10 +261,10 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente /** * Execute the business logic, delegating to the writer.
* - * Process the items with the {@link ItemWriter} in a stateful retry. - * Any {@link SkipListener} provided is called when retry attempts are - * exhausted. The listener callback (on write failure) will happen in - * the next transaction automatically.
+ * Process the items with the {@link ItemWriter} in a stateful retry. Any + * {@link SkipListener} provided is called when retry attempts are + * exhausted. The listener callback (on write failure) will happen in the + * next transaction automatically.
*/ protected void write(final Chunk chunk, final StepContribution contribution) throws Exception { @@ -324,7 +321,7 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente } }; - retryOperations.execute(retryCallback, recoveryCallback, new DefaultRetryState(chunk,rollbackClassifier)); + retryOperations.execute(retryCallback, recoveryCallback, new DefaultRetryState(chunk, rollbackClassifier)); for (ItemWrapper skip : chunk.getSkips()) { Exception exception = skip.getException(); @@ -340,7 +337,6 @@ public class FaultTolerantChunkOrientedTasklet extends AbstractItemOriente } - /** * @param attributes */