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

OPEN - BATCH-803: Add non-buffering ChunkOrientedTasklet (or option in existing one) plus flag for factory bean

Browse Source 
pulled generic item-oriented parts of ChunkOrientedTasklet into abstract superclass
This commit is contained in:
robokaso
2008-09-16 12:31:38 +00:00
parent 0ae4fa2461
commit 86e453b633
3 changed files with 126 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemProcessingTasklet.java Normal file
View File

@@ -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 <I> input item type
* @param <O> output item type
*/
public abstract class AbstractItemProcessingTasklet<I, O> implements Tasklet {
protected final Log logger = LogFactory.getLog(getClass());
protected final ItemReader<? extends I> itemReader;
protected final ItemProcessor<? super I, ? extends O> itemProcessor;
protected final ItemWriter<? super O> itemWriter;
protected final MulticasterBatchListener<I, O> listener = new MulticasterBatchListener<I, O>();
public AbstractItemProcessingTasklet(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> 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<O> items) throws Exception {
try {
listener.beforeWrite(items);
itemWriter.write(items);
listener.afterWrite(items);
}
catch (Exception e) {
listener.onWriteError(e, items);
throw e;
}
}
}

108
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java
View File

@@ -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<T, S> implements Tasklet {
public class ChunkOrientedTasklet<T, S> extends AbstractItemProcessingTasklet<T,S> {
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<? extends T> itemReader;
private final ItemProcessor<? super T, ? extends S> itemProcessor;
private final ItemWriter<? super S> itemWriter;
private final RepeatOperations repeatOperations;
final private MulticasterBatchListener<T,S> listener = new MulticasterBatchListener<T,S>();
/**
* @param itemReader
* @param itemProcessor
@@ -71,43 +55,10 @@ public class ChunkOrientedTasklet<T, S> implements Tasklet {
public ChunkOrientedTasklet(ItemReader<? extends T> itemReader,
ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> 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<T,S> 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<T, S> implements Tasklet {
return new ItemWrapper<T>(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<T, S> 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<S> 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<S> 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
*/

12
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipLimitStepFactoryBean.java
View File

@@ -347,7 +347,7 @@ public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S>
// 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<T, S> extends SimpleStepFactoryBean<T, S>
for (ItemWrapper<ItemWrapper<T>> 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<T, S> extends SimpleStepFactoryBean<T, S>
RetryCallback<Object> retryCallback = new RetryCallback<Object>() {
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<T, S> extends SimpleStepFactoryBean<T, S>
for (Chunk<S>.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<T, S> extends SimpleStepFactoryBean<T, S>
for (ItemWrapper<S> 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);