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

IN PROGRESS - BATCH-896: "DRY" FaultTolerantTasklet implementations

Browse Source
This commit is contained in:
robokaso
2008-10-31 09:27:47 +00:00
parent 05e3ea56b3
commit 1412135c0b
4 changed files with 106 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<I, O> extends AbstractItemOrientedTasklet<I, O> {
public AbstractFaultTolerantChunkOrientedTasklet(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends O> itemProcessor, ItemWriter<? super O> 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<Exception> skippedReads, final Map<I, Exception> skippedInputs,
final Map<O, Exception> skippedOutputs) {
for (Exception e : skippedReads) {
try {
listener.onSkipInRead(e);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
}
for (Entry<I, Exception> 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<O, Exception> 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 <T> List<T> getBufferedList(AttributeAccessor attributes, String key) {
List<T> buffer;
if (!attributes.hasAttribute(key)) {
buffer = new ArrayList<T>();
attributes.setAttribute(key, buffer);
}
else {
@SuppressWarnings("unchecked")
List<T> casted = (List<T>) 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 <T> Map<T, Exception> getBufferedSkips(AttributeAccessor attributes, String key) {
Map<T, Exception> buffer;
if (!attributes.hasAttribute(key)) {
buffer = new LinkedHashMap<T, Exception>();
attributes.setAttribute(key, buffer);
}
else {
@SuppressWarnings("unchecked")
Map<T, Exception> casted = (Map<T, Exception>) attributes.getAttribute(key);
buffer = casted;
}
return buffer;
}
}

79
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractItemOrientedTasklet.java
View File

@@ -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<I, O> 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<Exception> skippedReads, final Map<I, Exception> skippedInputs,
final Map<O, Exception> skippedOutputs) {
for (Exception e : skippedReads) {
try {
listener.onSkipInRead(e);
}
catch (RuntimeException ex) {
throw new SkipListenerFailedException("Fatal exception in SkipListener.", ex, e);
}
}
for (Entry<I, Exception> 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<O, Exception> 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 <T> List<T> getBufferedList(AttributeAccessor attributes, String key) {
List<T> buffer;
if (!attributes.hasAttribute(key)) {
buffer = new ArrayList<T>();
attributes.setAttribute(key, buffer);
}
else {
@SuppressWarnings("unchecked")
List<T> casted = (List<T>) 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 <T> Map<T, Exception> getBufferedSkips(AttributeAccessor attributes, String key) {
Map<T, Exception> buffer;
if (!attributes.hasAttribute(key)) {
buffer = new LinkedHashMap<T, Exception>();
attributes.setAttribute(key, buffer);
}
else {
@SuppressWarnings("unchecked")
Map<T, Exception> casted = (Map<T, Exception>) attributes.getAttribute(key);
buffer = casted;
}
return buffer;
}
}

20
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkOrientedTasklet.java
View File

@@ -55,7 +55,7 @@ import org.springframework.core.AttributeAccessor;
* @author Dave Syer
* @author Robert Kasanicky
*/
public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOrientedTasklet<T, S> {
public class FaultTolerantChunkOrientedTasklet<I, S> extends AbstractFaultTolerantChunkOrientedTasklet<I, S> {
private static final String INPUT_BUFFER_KEY = "INPUT_BUFFER_KEY";
@@ -77,8 +77,8 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
private static final String SKIPPED_READS_KEY = "SKIPPED_READS_BUFFER_KEY";
public FaultTolerantChunkOrientedTasklet(ItemReader<? extends T> itemReader,
ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter,
public FaultTolerantChunkOrientedTasklet(ItemReader<? extends I> itemReader,
ItemProcessor<? super I, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter,
RepeatOperations chunkOperations, RetryOperations retryTemplate,
Classifier<Throwable, Boolean> rollbackClassifier, ItemSkipPolicy readSkipPolicy,
ItemSkipPolicy writeSkipPolicy, ItemSkipPolicy processSkipPolicy) {
@@ -102,7 +102,7 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
*/
public ExitStatus execute(final StepContribution contribution, AttributeAccessor attributes) throws Exception {
final List<T> inputs = getBufferedList(attributes, INPUT_BUFFER_KEY);
final List<I> inputs = getBufferedList(attributes, INPUT_BUFFER_KEY);
final List<S> outputs = new ArrayList<S>();
ExitStatus result = ExitStatus.CONTINUABLE;
@@ -113,7 +113,7 @@ public class FaultTolerantChunkOrientedTasklet<T, S> 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<T, S> extends AbstractItemOriente
}
Map<T, Exception> skippedInputs = getBufferedSkips(attributes, SKIPPED_INPUTS_KEY);
Map<I, Exception> 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<T, S> extends AbstractItemOriente
* @param skippedReads
* @return next item for processing
*/
protected T read(StepContribution contribution, List<Exception> skippedReads) throws Exception {
protected I read(StepContribution contribution, List<Exception> skippedReads) throws Exception {
while (true) {
try {
@@ -200,12 +200,12 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
/**
* Incorporate retry into the item processor stage.
*/
protected void process(final StepContribution contribution, final List<T> inputs, final List<S> outputs,
final Map<T, Exception> skippedInputs) throws Exception {
protected void process(final StepContribution contribution, final List<I> inputs, final List<S> outputs,
final Map<I, Exception> skippedInputs) throws Exception {
int filtered = 0;
for (final T item : inputs) {
for (final I item : inputs) {
RetryCallback<S> retryCallback = new RetryCallback<S>() {

2
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/NonbufferingFaultTolerantChunkOrientedTasklet.java
View File

@@ -39,7 +39,7 @@ import org.springframework.core.AttributeAccessor;
* @param <I> input item type
* @param <O> output item type
*/
public class NonbufferingFaultTolerantChunkOrientedTasklet<I, O> extends AbstractItemOrientedTasklet<I, O> {
public class NonbufferingFaultTolerantChunkOrientedTasklet<I, O> extends AbstractFaultTolerantChunkOrientedTasklet<I, O> {
private static final String SKIPPED_INPUTS_KEY = "SKIPPED_INPUTS_KEY";