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 
inlined short single-use helper methods to make the execution logic more obvious
This commit is contained in:
robokaso
2008-09-17 12:06:46 +00:00
parent c4291f10af
commit bfa7e2a8ea
1 changed files with 14 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -95,8 +95,8 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
// TODO: check flags to see if these need to be saved or not (e.g. JMS
// not)
final Chunk<T> inputs = getInputBuffer(attributes);
final Chunk<S> outputs = getOutputBuffer(attributes);
final Chunk<T> inputs = getBuffer(attributes, INPUT_BUFFER_KEY);
final Chunk<S> outputs = getBuffer(attributes, OUTPUT_BUFFER_KEY);
ExitStatus result = ExitStatus.CONTINUABLE;
@@ -120,7 +120,8 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
return result;
}
storeInputs(attributes, inputs);
// store inputs
attributes.setAttribute(INPUT_BUFFER_KEY, inputs);
}
@@ -128,7 +129,13 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
process(contribution, inputs, outputs);
}
storeOutputsAndClearInputs(attributes, outputs, contribution);
/* Savepoint at end of processing and before writing. The processed items
* ready for output are stored so that if writing fails they can be picked
* up again in the next try. The inputs are finished with so we can clear
* their attribute.
*/
attributes.setAttribute(OUTPUT_BUFFER_KEY, outputs);
attributes.removeAttribute(INPUT_BUFFER_KEY);
// TODO: use ItemWriter interface properly
// TODO: make sure exceptions get handled by the appropriate handler
@@ -137,7 +144,9 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
// On successful completion clear the attributes to signal that there is
// no more processing
if (outputs.isEmpty()) {
clearAll(attributes);
for (String key : attributes.attributeNames()) {
attributes.removeAttribute(key);
}
}
return result;
@@ -337,59 +346,6 @@ public class FaultTolerantChunkOrientedTasklet<T, S> extends AbstractItemOriente
}
/**
* @param attributes
*/
private static void clearInputs(AttributeAccessor attributes) {
attributes.removeAttribute(INPUT_BUFFER_KEY);
}
/**
* @param attributes
* @param inputs
*/
private void storeInputs(AttributeAccessor attributes, Chunk<T> inputs) {
store(attributes, INPUT_BUFFER_KEY, inputs);
}
/**
* Savepoint at end of processing and before writing. The processed items
* ready for output are stored so that if writing fails they can be picked
* up again in the next try. The inputs are finished with so we can clear
* their attribute.
*
* @param attributes
* @param outputs
*/
private void storeOutputsAndClearInputs(AttributeAccessor attributes, Chunk<S> outputs,
StepContribution contribution) {
store(attributes, OUTPUT_BUFFER_KEY, outputs);
clearInputs(attributes);
}
/**
* @param attributes
* @param inputBufferKey
* @param outputs
*/
private static <W> void store(AttributeAccessor attributes, String key, W value) {
attributes.setAttribute(key, value);
}
private static void clearAll(AttributeAccessor attributes) {
for (String key : attributes.attributeNames()) {
attributes.removeAttribute(key);
}
}
private Chunk<T> getInputBuffer(AttributeAccessor attributes) {
return getBuffer(attributes, INPUT_BUFFER_KEY);
}
private Chunk<S> getOutputBuffer(AttributeAccessor attributes) {
return getBuffer(attributes, OUTPUT_BUFFER_KEY);
}
/**
* @param attributes
* @param inputBufferKey