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

BATCH-365: Moved reset and mark to ItemReader interface.

Browse Source
This commit is contained in:
lucasward
2008-02-24 21:14:51 +00:00
parent 747a706e86
commit fb98433034
9 changed files with 33 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemOrientedStep.java
View File

@@ -279,7 +279,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
TransactionStatus transaction = streamManager.getTransaction(stepExecution);
try {
itemReader.mark();
result = processChunk(contribution);
// TODO: check that stepExecution can
@@ -327,6 +327,7 @@ public class ItemOrientedStep extends AbstractStep implements InitializingBean {
stepExecution.rollback();
}
try {
itemReader.reset();
itemWriter.clear();
streamManager.rollback(transaction);
}

24
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemChunker.java
View File

@@ -27,8 +27,6 @@ import org.springframework.batch.io.exception.ReadFailureException;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.exception.MarkFailedException;
import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.item.exception.StreamException;
import org.springframework.util.Assert;
@@ -59,6 +57,7 @@ public class ItemChunker implements Chunker {
public ChunkingResult chunk(int size, StepContribution stepContribution) throws ReadFailureException {
Assert.isTrue(size > 0, "Chunk size must be greater than 0");
itemReader.mark();
int counter = 0;
List items = new ArrayList(size);
List exceptions = new ArrayList();
@@ -75,6 +74,7 @@ public class ItemChunker implements Chunker {
} catch (Exception ex) {
exceptions.add(ex);
if(!itemSkipPolicy.shouldSkip(ex, stepContribution)){
itemReader.reset();
rethrow(ex);
}
}
@@ -108,32 +108,12 @@ public class ItemChunker implements Chunker {
}
}
public boolean isMarkSupported() {
if(itemReader instanceof ItemStream){
return ((ItemStream)itemReader).isMarkSupported();
}
return false;
}
public void mark() throws MarkFailedException {
if(itemReader instanceof ItemStream){
((ItemStream)itemReader).mark();
}
}
public void open() throws StreamException {
if(itemReader instanceof ItemStream){
((ItemStream)itemReader).open();
}
}
public void reset() throws ResetFailedException {
if(itemReader instanceof ItemStream){
((ItemStream)itemReader).reset();
}
}
public void restoreFrom(ExecutionContext context) {
if(itemReader instanceof ItemStream){
((ItemStream)itemReader).restoreFrom(context);

25
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemDechunker.java
View File

@@ -20,16 +20,14 @@ import java.util.Iterator;
import java.util.List;
import org.springframework.batch.core.domain.Chunk;
import org.springframework.batch.core.domain.DechunkingResult;
import org.springframework.batch.core.domain.Dechunker;
import org.springframework.batch.core.domain.DechunkingResult;
import org.springframework.batch.core.domain.ItemSkipPolicy;
import org.springframework.batch.core.domain.StepContribution;
import org.springframework.batch.io.exception.WriteFailureException;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.exception.MarkFailedException;
import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.item.exception.StreamException;
import org.springframework.util.Assert;
@@ -99,33 +97,12 @@ public class ItemDechunker implements Dechunker {
}
}
public boolean isMarkSupported() {
if(itemWriter instanceof ItemStream){
return ((ItemStream)itemWriter).isMarkSupported();
}
else{
return false;
}
}
public void mark() throws MarkFailedException {
if(itemWriter instanceof ItemStream){
((ItemStream)itemWriter).mark();
}
}
public void open() throws StreamException {
if(itemWriter instanceof ItemStream){
((ItemStream)itemWriter).open();
}
}
public void reset() throws ResetFailedException {
if(itemWriter instanceof ItemStream){
((ItemStream)itemWriter).reset();
}
}
public void restoreFrom(ExecutionContext context) {
if(itemWriter instanceof ItemStream){
((ItemStream)itemWriter).restoreFrom(context);

7
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ChunkedStepTests.java
View File

@@ -36,6 +36,7 @@ import org.springframework.batch.execution.step.support.ItemDechunker;
import org.springframework.batch.execution.step.support.JobRepositorySupport;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.reader.AbstractItemReader;
import org.springframework.batch.item.reader.ListItemReader;
import org.springframework.batch.item.stream.SimpleStreamManager;
import org.springframework.batch.item.writer.AbstractItemWriter;
@@ -118,7 +119,7 @@ public class ChunkedStepTests extends TestCase {
final JobExecution jobExecution = new JobExecution(jobInstance);
final StepExecution stepExecution = new StepExecution("testStep", jobExecution);
chunkedStep.setChunker(new ItemChunker(new ItemReader() {
chunkedStep.setChunker(new ItemChunker(new AbstractItemReader() {
int counter = 0;
public Object read() throws Exception {
assertNotNull(StepSynchronizationManager.getContext().getStepExecution());
@@ -192,7 +193,7 @@ public class ChunkedStepTests extends TestCase {
//fail.
public void testReadFailure() {
ItemReader itemReader = new ItemReader() {
ItemReader itemReader = new AbstractItemReader() {
int counter = 0;
public Object read() throws Exception {
@@ -242,7 +243,7 @@ public class ChunkedStepTests extends TestCase {
public void testExitCodeDefaultClassification() throws Exception {
ItemReader itemReader = new ItemReader() {
ItemReader itemReader = new AbstractItemReader() {
int counter = 0;
public Object read() throws Exception {
counter++;

15
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/ItemOrientedStepTests.java
View File

@@ -46,6 +46,7 @@ import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.item.exception.StreamException;
import org.springframework.batch.item.reader.AbstractItemReader;
import org.springframework.batch.item.reader.ListItemReader;
import org.springframework.batch.item.stream.ItemStreamAdapter;
import org.springframework.batch.item.stream.SimpleStreamManager;
@@ -159,7 +160,7 @@ public class ItemOrientedStepTests extends TestCase {
final JobExecution jobExecution = new JobExecution(jobInstance);
final StepExecution stepExecution = new StepExecution(step, jobExecution);
itemOrientedStep.setItemReader(new ItemReader() {
itemOrientedStep.setItemReader(new AbstractItemReader() {
public Object read() throws Exception {
assertEquals(step, stepExecution.getStepName());
assertNotNull(StepSynchronizationManager.getContext().getStepExecution());
@@ -214,7 +215,7 @@ public class ItemOrientedStepTests extends TestCase {
public void testIncrementRollbackCount() {
ItemReader itemReader = new ItemReader() {
ItemReader itemReader = new AbstractItemReader() {
public Object read() throws Exception {
int counter = 0;
@@ -245,7 +246,7 @@ public class ItemOrientedStepTests extends TestCase {
public void testExitCodeDefaultClassification() throws Exception {
ItemReader itemReader = new ItemReader() {
ItemReader itemReader = new AbstractItemReader() {
public Object read() throws Exception {
int counter = 0;
@@ -348,7 +349,7 @@ public class ItemOrientedStepTests extends TestCase {
public void testRestartJobOnNonRestartableTasklet() throws Exception {
String step = "stepName";
// step.setStepExecutionCount(1);
itemOrientedStep.setItemReader(new ItemReader() {
itemOrientedStep.setItemReader(new AbstractItemReader() {
public Object read() throws Exception {
return "foo";
}
@@ -402,7 +403,7 @@ public class ItemOrientedStepTests extends TestCase {
public void testStreamManager() throws Exception {
String step = "stepName";
// step.setStepExecutionCount(1);
itemOrientedStep.setItemReader(new ItemReader() {
itemOrientedStep.setItemReader(new AbstractItemReader() {
public Object read() throws Exception {
return "foo";
}
@@ -484,7 +485,7 @@ public class ItemOrientedStepTests extends TestCase {
itemOrientedStep.setInterruptionPolicy(interruptionPolicy);
ItemReader itemReader = new ItemReader() {
ItemReader itemReader = new AbstractItemReader() {
public Object read() throws Exception {
int counter = 0;
@@ -523,7 +524,7 @@ public class ItemOrientedStepTests extends TestCase {
public void testStatusForResetFailedException() throws Exception {
ItemReader itemReader = new ItemReader() {
ItemReader itemReader = new AbstractItemReader() {
public Object read() throws Exception {
// Trigger a rollback
throw new RuntimeException("Foo");

3
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/ItemDechunkerTests.java
View File

@@ -64,6 +64,7 @@ public class ItemDechunkerTests extends TestCase {
itemWriter.write("1");
itemWriter.write("2");
itemWriter.flush();
writerControl.replay();
dechunker.dechunk(chunk, stepContribution);
writerControl.verify();
@@ -75,6 +76,7 @@ public class ItemDechunkerTests extends TestCase {
itemWriter.write("1");
itemWriter.write("2");
writerControl.setThrowable(new Exception());
itemWriter.flush();
writerControl.replay();
DechunkingResult result = dechunker.dechunk(chunk, stepContribution);
writerControl.verify();
@@ -89,6 +91,7 @@ public class ItemDechunkerTests extends TestCase {
itemWriter.write("1");
itemWriter.write("2");
writerControl.setThrowable(new NullPointerException());
itemWriter.clear();
writerControl.replay();
try{
dechunker.dechunk(chunk, stepContribution);

8
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/MockItemReader.java
View File

@@ -16,6 +16,8 @@
package org.springframework.batch.execution.step.support;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.exception.MarkFailedException;
import org.springframework.batch.item.exception.ResetFailedException;
public class MockItemReader implements ItemReader {
@@ -57,4 +59,10 @@ public class MockItemReader implements ItemReader {
return null;
}
public void mark() throws MarkFailedException {
}
public void reset() throws ResetFailedException {
}
}

6
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/RepeatOperationsStepTests.java
View File

@@ -25,7 +25,7 @@ import org.springframework.batch.core.domain.JobInstance;
import org.springframework.batch.core.domain.JobParameters;
import org.springframework.batch.core.domain.StepExecution;
import org.springframework.batch.execution.step.ItemOrientedStep;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.reader.AbstractItemReader;
import org.springframework.batch.item.reader.ItemReaderAdapter;
import org.springframework.batch.item.writer.AbstractItemWriter;
import org.springframework.batch.item.writer.ItemWriterAdapter;
@@ -82,7 +82,7 @@ public class RepeatOperationsStepTests extends TestCase {
});
repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(2));
ItemOrientedStep configuration = new ItemOrientedStep();
configuration.setItemReader(new ItemReader(){
configuration.setItemReader(new AbstractItemReader(){
public Object read() throws Exception {
throw new NullPointerException();
}});
@@ -122,7 +122,7 @@ public class RepeatOperationsStepTests extends TestCase {
});
stepTemplate.setCompletionPolicy(new SimpleCompletionPolicy(1));
RepeatOperationsStep configuration = new RepeatOperationsStep();
configuration.setItemReader(new ItemReader(){
configuration.setItemReader(new AbstractItemReader(){
public Object read() throws Exception {
return new Object();
}});

4
spring-batch-execution/src/test/java/org/springframework/batch/execution/step/support/StepExecutorInterruptionTests.java
View File

@@ -34,7 +34,7 @@ import org.springframework.batch.execution.repository.dao.JobInstanceDao;
import org.springframework.batch.execution.repository.dao.MapJobDao;
import org.springframework.batch.execution.repository.dao.MapStepDao;
import org.springframework.batch.execution.repository.dao.StepExecutionDao;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.reader.AbstractItemReader;
import org.springframework.batch.item.reader.ItemReaderAdapter;
import org.springframework.batch.item.writer.AbstractItemWriter;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
@@ -80,7 +80,7 @@ public class StepExecutorInterruptionTests extends TestCase {
final String stepName = (String) steps.get(0);
JobExecution jobExecutionContext = new JobExecution(new JobInstance(new Long(0L), new JobParameters()));
final StepExecution stepExecution = new StepExecution(stepName, jobExecutionContext);
step.setItemReader(new ItemReader() {
step.setItemReader(new AbstractItemReader() {
public Object read() throws Exception {
// do something non-trivial (and not Thread.sleep())
double foo = 1;