From 86e453b633d5fb3ecde6cec881680eab32803dea Mon Sep 17 00:00:00 2001 From: robokaso Date: Tue, 16 Sep 2008 12:31:38 +0000 Subject: [PATCH] OPEN - BATCH-803: Add non-buffering ChunkOrientedTasklet (or option in existing one) plus flag for factory bean pulled generic item-oriented parts of ChunkOrientedTasklet into abstract superclass --- .../item/AbstractItemProcessingTasklet.java | 115 ++++++++++++++++++ .../core/step/item/ChunkOrientedTasklet.java | 108 +--------------- .../step/item/SkipLimitStepFactoryBean.java | 12 +- 3 files changed, 126 insertions(+), 109 deletions(-) create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemProcessingTasklet.java diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemProcessingTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemProcessingTasklet.java new file mode 100644 index 000000000..93c37fa3d --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemProcessingTasklet.java @@ -0,0 +1,115 @@ +package org.springframework.batch.core.step.item; + +import java.util.List; + +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.tasklet.Tasklet; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; + +/** + * Superclass for {@link Tasklet}s implementing variations on read-process-write + * item handling. Encapsulates listener registration and bundles listener + * callbacks with relevant method calls. + * + * @author Robert Kasanicky + * + * @param input item type + * @param output item type + */ +public abstract class AbstractItemProcessingTasklet implements Tasklet { + + protected final Log logger = LogFactory.getLog(getClass()); + + protected final ItemReader itemReader; + + protected final ItemProcessor itemProcessor; + + protected final ItemWriter itemWriter; + + protected final MulticasterBatchListener listener = new MulticasterBatchListener(); + + public AbstractItemProcessingTasklet(ItemReader itemReader, + ItemProcessor itemProcessor, ItemWriter itemWriter) { + this.itemReader = itemReader; + this.itemProcessor = itemProcessor; + this.itemWriter = itemWriter; + } + + /** + * Register some {@link StepListener}s with the handler. Each will get the + * callbacks in the order specified at the correct stage. + * + * @param listeners + */ + public void setListeners(StepListener[] listeners) { + for (StepListener listener : listeners) { + registerListener(listener); + } + } + + /** + * Register a listener for callbacks at the appropriate stages in a process. + * + * @param listener a {@link StepListener} + */ + public void registerListener(StepListener listener) { + this.listener.register(listener); + } + + /** + * @return item + * @throws Exception + */ + protected final I doRead() throws Exception { + try { + listener.beforeRead(); + I item = itemReader.read(); + listener.afterRead(item); + return item; + } + catch (Exception e) { + listener.onReadError(e); + throw e; + } + } + + /** + * @param item the input item + * @return the result of the processing + * @throws Exception + */ + protected final O doProcess(I item) throws Exception { + try { + listener.beforeProcess(item); + O result = itemProcessor.process(item); + listener.afterProcess(item, result); + return result; + } + catch (Exception e) { + listener.onProcessError(item, e); + throw e; + } + } + + /** + * @param items + * @throws Exception + */ + protected final void doWrite(List items) throws Exception { + try { + listener.beforeWrite(items); + itemWriter.write(items); + listener.afterWrite(items); + } + catch (Exception e) { + listener.onWriteError(e, items); + throw e; + } + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java index cba070236..7580d7cf3 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java @@ -15,13 +15,7 @@ */ package org.springframework.batch.core.step.item; -import java.util.List; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.StepContribution; -import org.springframework.batch.core.StepListener; -import org.springframework.batch.core.listener.MulticasterBatchListener; import org.springframework.batch.core.step.tasklet.Tasklet; import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemReader; @@ -44,24 +38,14 @@ import org.springframework.core.AttributeAccessor; * @author Dave Syer * @author Robert Kasanicky */ -public class ChunkOrientedTasklet implements Tasklet { +public class ChunkOrientedTasklet extends AbstractItemProcessingTasklet { private static final String INPUT_BUFFER_KEY = "INPUT_BUFFER_KEY"; private static final String OUTPUT_BUFFER_KEY = "OUTPUT_BUFFER_KEY"; - protected final Log logger = LogFactory.getLog(getClass()); - - private final ItemReader itemReader; - - private final ItemProcessor itemProcessor; - - private final ItemWriter itemWriter; - private final RepeatOperations repeatOperations; - final private MulticasterBatchListener listener = new MulticasterBatchListener(); - /** * @param itemReader * @param itemProcessor @@ -71,43 +55,10 @@ public class ChunkOrientedTasklet implements Tasklet { public ChunkOrientedTasklet(ItemReader itemReader, ItemProcessor itemProcessor, ItemWriter itemWriter, RepeatOperations repeatOperations) { - super(); - this.itemReader = itemReader; - this.itemProcessor = itemProcessor; - this.itemWriter = itemWriter; + super(itemReader, itemProcessor, itemWriter); this.repeatOperations = repeatOperations; } - /** - * Register some {@link StepListener}s with the handler. Each will get - * the callbacks in the order specified at the correct stage. - * - * @param listeners - */ - public void setListeners(StepListener[] listeners) { - for (StepListener listener : listeners) { - registerListener(listener); - } - } - - /** - * Register a listener for callbacks at the appropriate stages in a - * process. - * - * @param listener a {@link StepListener} - */ - public void registerListener(StepListener listener) { - this.listener.register(listener); - } - - /** - * Public getter for the listener. - * @return the listener - */ - protected MulticasterBatchListener getListener() { - return listener; - } - /** * Get the next item from {@link #read(StepContribution)} and if not null * pass the item to {@link #write(Chunk, StepContribution)}. If the @@ -177,23 +128,6 @@ public class ChunkOrientedTasklet implements Tasklet { return new ItemWrapper(doRead()); } - /** - * @return item - * @throws Exception - */ - protected final T doRead() throws Exception { - try { - listener.beforeRead(); - T item = itemReader.read(); - listener.afterRead(item); - return item; - } - catch (Exception e) { - listener.onReadError(e); - throw e; - } - } - /** * * @param inputs the items to process @@ -215,51 +149,17 @@ public class ChunkOrientedTasklet implements Tasklet { inputs.clear(); } - /** - * @param item the input item - * @return the result of the processing - * @throws Exception - */ - protected S doProcess(T item) throws Exception { - try { - listener.beforeProcess(item); - S result = itemProcessor.process(item); - listener.afterProcess(item, result); - return result; - } - catch (Exception e) { - listener.onProcessError(item, e); - throw e; - } - } - /** * * @param chunk the items to write * @param contribution current context */ protected void write(Chunk chunk, StepContribution contribution) throws Exception { - doWrite(chunk.getItems(), contribution); + doWrite(chunk.getItems()); + contribution.incrementWriteCount(chunk.size()); chunk.clear(); } - /** - * @param items - * @throws Exception - */ - protected final void doWrite(List items, StepContribution contribution) throws Exception { - try { - listener.beforeWrite(items); - itemWriter.write(items); - contribution.incrementWriteCount(items.size()); - listener.afterWrite(items); - } - catch (Exception e) { - listener.onWriteError(e, items); - throw e; - } - } - /** * @param attributes */ diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java index c60b1e841..6e537f969 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java @@ -347,7 +347,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean // increment skip count and try again try { skipCount++; - getListener().onSkipInRead(e); + listener.onSkipInRead(e); } catch (RuntimeException ex) { contribution.incrementReadSkipCount(skipCount); @@ -427,7 +427,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean for (ItemWrapper> skip : inputs.getSkips()) { Exception exception = skip.getException(); try { - getListener().onSkipInProcess(skip.getItem().getItem(), exception); + listener.onSkipInProcess(skip.getItem().getItem(), exception); } catch (RuntimeException e) { throw new SkipListenerFailedException("Fatal exception in SkipListener.", e, exception); @@ -453,7 +453,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean RetryCallback retryCallback = new RetryCallback() { public Object doWithRetry(RetryContext context) throws Exception { - doWrite(chunk.getItems(), contribution); + doWrite(chunk.getItems()); + contribution.incrementWriteCount(chunk.size()); return null; } }; @@ -473,7 +474,8 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean for (Chunk.ChunkIterator iterator = chunk.iterator(); iterator.hasNext();) { S item = iterator.next(); try { - doWrite(Collections.singletonList(item), contribution); + doWrite(Collections.singletonList(item)); + contribution.incrementWriteCount(1); } catch (Exception e) { checkSkipPolicy(contribution, iterator, e); @@ -507,7 +509,7 @@ public class SkipLimitStepFactoryBean extends SimpleStepFactoryBean for (ItemWrapper skip : chunk.getSkips()) { Exception exception = skip.getException(); try { - getListener().onSkipInWrite(skip.getItem(), exception); + listener.onSkipInWrite(skip.getItem(), exception); } catch (RuntimeException e) { throw new SkipListenerFailedException("Fatal exception in SkipListener.", e, exception);