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

RESOLVED - issue BATCH-404: FactoryBeans for step configuration

Browse Source 
http://jira.springframework.org/browse/BATCH-404

ItemProcessor -> ItemHandler.
This commit is contained in:
dsyer
2008-03-03 18:01:02 +00:00
parent 36801d7d5d
commit 379f2fe826
12 changed files with 89 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemProcessor.java → spring-batch-execution/src/main/java/org/springframework/batch/execution/step/ItemHandler.java
View File

@@ -18,6 +18,10 @@ package org.springframework.batch.execution.step;
import org.springframework.batch.core.domain.StepContribution;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.exception.ClearFailedException;
import org.springframework.batch.item.exception.FlushFailedException;
import org.springframework.batch.item.exception.MarkFailedException;
import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.repeat.ExitStatus;
/**
@@ -28,7 +32,7 @@ import org.springframework.batch.repeat.ExitStatus;
* @author Dave Syer
*
*/
public interface ItemProcessor extends ItemReader, ItemWriter {
public interface ItemHandler {
/**
* Given the current context in the form of a step contribution, do whatever
@@ -41,6 +45,34 @@ public interface ItemProcessor extends ItemReader, ItemWriter {
* @return an {@link ExitStatus} indicating whether processing is
* continuable.
*/
ExitStatus process(StepContribution contribution) throws Exception;
ExitStatus handle(StepContribution contribution) throws Exception;
/**
* Implementations should delegate to an {@link ItemReader}.
*
* @see org.springframework.batch.item.ItemReader#mark()
*/
void mark() throws MarkFailedException;
/**
* Implementations should delegate to an {@link ItemReader}.
*
* @see org.springframework.batch.item.ItemReader#reset()
*/
void reset() throws ResetFailedException;
/**
* Implementations should delegate to an {@link ItemWriter}.
*
* @see org.springframework.batch.item.ItemWriter#flush()
*/
public void flush() throws FlushFailedException;
/**
* Implementations should delegate to an {@link ItemWriter}.
*
* @see org.springframework.batch.item.ItemWriter#clear()
*/
public void clear() throws ClearFailedException;
}

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

@@ -89,7 +89,7 @@ public class ItemOrientedStep extends AbstractStep {
private PlatformTransactionManager transactionManager;
private ItemProcessor itemProcessor;
private ItemHandler itemHandler;
/**
* @param name
@@ -117,11 +117,11 @@ public class ItemOrientedStep extends AbstractStep {
}
/**
* Public setter for the {@link ItemProcessor}.
* @param itemProcessor the {@link ItemProcessor} to set
* Public setter for the {@link ItemHandler}.
* @param itemHandler the {@link ItemHandler} to set
*/
public void setItemProcessor(ItemProcessor itemProcessor) {
this.itemProcessor = itemProcessor;
public void setItemProcessor(ItemHandler itemHandler) {
this.itemHandler = itemHandler;
}
/**
@@ -281,7 +281,7 @@ public class ItemOrientedStep extends AbstractStep {
try {
itemProcessor.mark();
itemHandler.mark();
result = processChunk(contribution);
contribution.incrementCommitCount();
@@ -307,8 +307,8 @@ public class ItemOrientedStep extends AbstractStep {
}
try {
itemProcessor.mark();
itemProcessor.flush();
itemHandler.mark();
itemHandler.flush();
transactionManager.commit(transaction);
}
catch (Exception e) {
@@ -334,8 +334,8 @@ public class ItemOrientedStep extends AbstractStep {
}
try {
itemProcessor.reset();
itemProcessor.clear();
itemHandler.reset();
itemHandler.clear();
transactionManager.rollback(transaction);
}
catch (Exception e) {
@@ -457,7 +457,7 @@ public class ItemOrientedStep extends AbstractStep {
}
// check for interruption before each item as well
interruptionPolicy.checkInterrupted(context);
ExitStatus exitStatus = itemProcessor.process(contribution);
ExitStatus exitStatus = itemHandler.handle(contribution);
contribution.incrementTaskCount();
// check for interruption after each item as well
interruptionPolicy.checkInterrupted(context);

2
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/AbstractStepFactoryBean.java
View File

@@ -157,7 +157,7 @@ public abstract class AbstractStepFactoryBean extends AbstractFactoryBean implem
Assert.notNull(jobRepository, "JobRepository must be provided");
Assert.notNull(transactionManager, "TransactionManager must be provided");
step.setItemProcessor(new SimpleItemProcessor(itemReader, itemWriter));
step.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
step.setTransactionManager(transactionManager);
step.setJobRepository(jobRepository);
step.setStartLimit(startLimit);

16
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/DefaultStepFactoryBean.java
View File

@@ -19,7 +19,7 @@ import org.springframework.batch.core.domain.BatchListener;
import org.springframework.batch.core.domain.Step;
import org.springframework.batch.core.domain.StepListener;
import org.springframework.batch.execution.step.ItemOrientedStep;
import org.springframework.batch.execution.step.ItemProcessor;
import org.springframework.batch.execution.step.ItemHandler;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
@@ -50,7 +50,7 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean {
private TaskExecutor taskExecutor;
private ItemProcessor itemProcessor;
private ItemHandler itemHandler;
private RepeatTemplate stepOperations;
@@ -121,16 +121,16 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean {
* Public getter for the ItemProcessor.
* @return the itemProcessor
*/
protected ItemProcessor getItemProcessor() {
return itemProcessor;
protected ItemHandler getItemProcessor() {
return itemHandler;
}
/**
* Public setter for the ItemProcessor.
* @param itemProcessor the itemProcessor to set
* @param itemHandler the itemProcessor to set
*/
protected void setItemProcessor(ItemProcessor itemProcessor) {
this.itemProcessor = itemProcessor;
protected void setItemProcessor(ItemHandler itemHandler) {
this.itemHandler = itemHandler;
}
/**
@@ -200,7 +200,7 @@ public class DefaultStepFactoryBean extends AbstractStepFactoryBean {
step.setStepOperations(stepOperations);
ItemSkipPolicyItemProcessor itemProcessor = new ItemSkipPolicyItemProcessor(itemReader, itemWriter);
ItemSkipPolicyItemHandler itemProcessor = new ItemSkipPolicyItemHandler(itemReader, itemWriter);
if (skipLimit > 0) {
/*

8
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemSkipPolicyItemProcessor.java → spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/ItemSkipPolicyItemHandler.java
View File

@@ -26,7 +26,7 @@ import org.springframework.batch.repeat.ExitStatus;
* @author Dave Syer
*
*/
public class ItemSkipPolicyItemProcessor extends SimpleItemProcessor {
public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy();
@@ -34,7 +34,7 @@ public class ItemSkipPolicyItemProcessor extends SimpleItemProcessor {
* @param itemReader
* @param itemWriter
*/
public ItemSkipPolicyItemProcessor(ItemReader itemReader, ItemWriter itemWriter) {
public ItemSkipPolicyItemHandler(ItemReader itemReader, ItemWriter itemWriter) {
super(itemReader, itemWriter);
}
@@ -60,12 +60,12 @@ public class ItemSkipPolicyItemProcessor extends SimpleItemProcessor {
* @return {@link ExitStatus#CONTINUABLE} if there is more processing to do
* @throws Exception if there is an error
*/
public ExitStatus process(StepContribution contribution) throws Exception {
public ExitStatus handle(StepContribution contribution) throws Exception {
ExitStatus exitStatus = ExitStatus.CONTINUABLE;
try {
exitStatus = super.process(contribution);
exitStatus = super.handle(contribution);
}
catch (Exception e) {

2
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/RepeatOperationsStepFactoryBean.java
View File

@@ -130,7 +130,7 @@ public class RepeatOperationsStepFactoryBean extends AbstractStepFactoryBean {
setItemWriter(itemWriter);
step.setStepListeners(stepListeners);
step.setItemProcessor(new SimpleItemProcessor(itemReader, itemWriter));
step.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
step.setChunkOperations(chunkOperations);
step.setStepOperations(stepOperations);

34
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SimpleItemProcessor.java → spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/SimpleItemHandler.java
View File

@@ -16,7 +16,7 @@
package org.springframework.batch.execution.step.support;
import org.springframework.batch.core.domain.StepContribution;
import org.springframework.batch.execution.step.ItemProcessor;
import org.springframework.batch.execution.step.ItemHandler;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.exception.ClearFailedException;
@@ -26,14 +26,14 @@ import org.springframework.batch.item.exception.ResetFailedException;
import org.springframework.batch.repeat.ExitStatus;
/**
* Simplest possible implementation of {@link ItemProcessor} with no skipping or
* Simplest possible implementation of {@link ItemHandler} with no skipping or
* recovering. Just delegates all calls to the provided {@link ItemReader} and
* {@link ItemWriter}.
*
* @author Dave Syer
*
*/
public class SimpleItemProcessor implements ItemProcessor {
public class SimpleItemHandler implements ItemHandler {
private ItemReader itemReader;
@@ -43,7 +43,7 @@ public class SimpleItemProcessor implements ItemProcessor {
* @param itemReader
* @param itemWriter
*/
public SimpleItemProcessor(ItemReader itemReader, ItemWriter itemWriter) {
public SimpleItemHandler(ItemReader itemReader, ItemWriter itemWriter) {
super();
this.itemReader = itemReader;
this.itemWriter = itemWriter;
@@ -69,14 +69,14 @@ public class SimpleItemProcessor implements ItemProcessor {
* Read from the {@link ItemReader} and process (if not null) with the
* {@link ItemWriter}.
*
* @see org.springframework.batch.execution.step.ItemProcessor#process(org.springframework.batch.core.domain.StepContribution)
* @see org.springframework.batch.execution.step.ItemHandler#handle(org.springframework.batch.core.domain.StepContribution)
*/
public ExitStatus process(StepContribution contribution) throws Exception {
Object item = read();
public ExitStatus handle(StepContribution contribution) throws Exception {
Object item = itemReader.read();
if (item == null) {
return ExitStatus.FINISHED;
}
write(item);
itemWriter.write(item);
return ExitStatus.CONTINUABLE;
}
@@ -88,15 +88,6 @@ public class SimpleItemProcessor implements ItemProcessor {
itemReader.mark();
}
/**
* @return
* @throws Exception
* @see org.springframework.batch.item.ItemReader#read()
*/
public Object read() throws Exception {
return itemReader.read();
}
/**
* @throws ResetFailedException
* @see org.springframework.batch.item.ItemReader#reset()
@@ -121,13 +112,4 @@ public class SimpleItemProcessor implements ItemProcessor {
itemWriter.flush();
}
/**
* @param item
* @throws Exception
* @see org.springframework.batch.item.ItemWriter#write(java.lang.Object)
*/
public void write(Object item) throws Exception {
itemWriter.write(item);
}
}

8
spring-batch-execution/src/main/java/org/springframework/batch/execution/step/support/StatefulRetryStepFactoryBean.java
View File

@@ -94,7 +94,7 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setRetryPolicy(itemProviderRetryPolicy);
StatefulRetryItemProcessor itemProcessor = new StatefulRetryItemProcessor(getItemReader(), getItemWriter(), retryTemplate, retryCallback);
StatefulRetryItemHandler itemProcessor = new StatefulRetryItemHandler(getItemReader(), getItemWriter(), retryTemplate, retryCallback);
step.setItemProcessor(itemProcessor);
@@ -120,7 +120,7 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
}
private static class StatefulRetryItemProcessor extends SimpleItemProcessor {
private static class StatefulRetryItemHandler extends SimpleItemHandler {
final private RetryOperations retryOperations;
@@ -132,7 +132,7 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
* @param retryCallback
* @param retryTemplate
*/
public StatefulRetryItemProcessor(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate, ItemReaderRetryCallback retryCallback) {
public StatefulRetryItemHandler(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate, ItemReaderRetryCallback retryCallback) {
super(itemReader, itemWriter);
this.retryOperations = retryTemplate;
this.retryCallback = retryCallback;
@@ -159,7 +159,7 @@ public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
* do
* @throws Exception if there is an error
*/
public ExitStatus process(StepContribution contribution) throws Exception {
public ExitStatus handle(StepContribution contribution) throws Exception {
return new ExitStatus(retryOperations.execute(retryCallback) != null);
}

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

@@ -37,7 +37,7 @@ import org.springframework.batch.execution.repository.dao.MapJobExecutionDao;
import org.springframework.batch.execution.repository.dao.MapJobInstanceDao;
import org.springframework.batch.execution.repository.dao.MapStepExecutionDao;
import org.springframework.batch.execution.step.support.JobRepositorySupport;
import org.springframework.batch.execution.step.support.SimpleItemProcessor;
import org.springframework.batch.execution.step.support.SimpleItemHandler;
import org.springframework.batch.execution.step.support.StepInterruptionPolicy;
import org.springframework.batch.io.exception.InfrastructureException;
import org.springframework.batch.item.ExecutionContext;
@@ -86,7 +86,7 @@ public class ItemOrientedStepTests extends TestCase {
private AbstractStep getStep(String[] strings) throws Exception {
ItemOrientedStep step = new ItemOrientedStep("stepName");
step.setItemProcessor(new SimpleItemProcessor(getReader(strings), itemWriter));
step.setItemProcessor(new SimpleItemHandler(getReader(strings), itemWriter));
step.setJobRepository(new JobRepositorySupport());
step.setTransactionManager(transactionManager);
return step;
@@ -168,7 +168,7 @@ public class ItemOrientedStepTests extends TestCase {
};
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(itemReader, itemWriter));
itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
JobExecution jobExecutionContext = new JobExecution(jobInstance);
StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
@@ -198,7 +198,7 @@ public class ItemOrientedStepTests extends TestCase {
};
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(itemReader, itemWriter));
itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
JobExecution jobExecutionContext = new JobExecution(jobInstance);
StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
@@ -217,7 +217,7 @@ public class ItemOrientedStepTests extends TestCase {
*/
public void testNonRestartedJob() throws Exception {
MockRestartableItemReader tasklet = new MockRestartableItemReader();
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(tasklet, itemWriter));
itemOrientedStep.setItemProcessor(new SimpleItemHandler(tasklet, itemWriter));
itemOrientedStep.registerStream(tasklet);
JobExecution jobExecutionContext = new JobExecution(jobInstance);
StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
@@ -288,7 +288,7 @@ public class ItemOrientedStepTests extends TestCase {
*/
public void testNoSaveExecutionAttributesRestartableJob() {
MockRestartableItemReader tasklet = new MockRestartableItemReader();
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(tasklet, itemWriter));
itemOrientedStep.setItemProcessor(new SimpleItemHandler(tasklet, itemWriter));
JobExecution jobExecutionContext = new JobExecution(jobInstance);
StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
@@ -308,7 +308,7 @@ public class ItemOrientedStepTests extends TestCase {
* Restartable.
*/
public void testRestartJobOnNonRestartableTasklet() throws Exception {
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(new AbstractItemReader() {
itemOrientedStep.setItemProcessor(new SimpleItemHandler(new AbstractItemReader() {
public Object read() throws Exception {
return "foo";
}
@@ -329,7 +329,7 @@ public class ItemOrientedStepTests extends TestCase {
executionContext.putString("foo", "bar");
}
};
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(reader, itemWriter));
itemOrientedStep.setItemProcessor(new SimpleItemHandler(reader, itemWriter));
itemOrientedStep.registerStream(reader);
JobExecution jobExecution = new JobExecution(jobInstance);
StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution);
@@ -422,7 +422,7 @@ public class ItemOrientedStepTests extends TestCase {
return null;
}
});
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(new MockRestartableItemReader() {
itemOrientedStep.setItemProcessor(new SimpleItemHandler(new MockRestartableItemReader() {
public Object read() throws Exception {
throw new RuntimeException("FOO");
}
@@ -448,7 +448,7 @@ public class ItemOrientedStepTests extends TestCase {
executionContext.putString("foo", "bar");
}
};
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(reader, itemWriter));
itemOrientedStep.setItemProcessor(new SimpleItemHandler(reader, itemWriter));
itemOrientedStep.setStreams(new ItemStream[] {reader});
JobExecution jobExecution = new JobExecution(jobInstance);
StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecution);
@@ -488,7 +488,7 @@ public class ItemOrientedStepTests extends TestCase {
};
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(itemReader, itemWriter));
itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
JobExecution jobExecutionContext = new JobExecution(jobInstance);
StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
@@ -516,7 +516,7 @@ public class ItemOrientedStepTests extends TestCase {
throw new RuntimeException("Foo");
}
};
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(itemReader, itemWriter));
itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
JobExecution jobExecutionContext = jobInstance.createJobExecution();
StepExecution stepExecution = new StepExecution(itemOrientedStep, jobExecutionContext);
@@ -543,7 +543,7 @@ public class ItemOrientedStepTests extends TestCase {
throw new RuntimeException("Foo");
}
};
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(itemReader, itemWriter));
itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
itemOrientedStep.setTransactionManager(new ResourcelessTransactionManager() {
protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
// Simulate failure on rollback when stream resets
@@ -639,7 +639,7 @@ public class ItemOrientedStepTests extends TestCase {
throw new RuntimeException("Bar");
}
};
itemOrientedStep.setItemProcessor(new SimpleItemProcessor(itemReader, itemWriter));
itemOrientedStep.setItemProcessor(new SimpleItemHandler(itemReader, itemWriter));
itemOrientedStep.registerStream(itemReader);
JobExecution jobExecutionContext = jobInstance.createJobExecution();

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

@@ -74,13 +74,13 @@ public class StepExecutorInterruptionTests extends TestCase {
public void write(Object item) throws Exception {
}
};
step.setItemProcessor(new SimpleItemProcessor(new ItemReaderAdapter(), itemWriter));
step.setItemProcessor(new SimpleItemHandler(new ItemReaderAdapter(), itemWriter));
}
public void testInterruptChunk() throws Exception {
final StepExecution stepExecution = new StepExecution(step, jobExecution);
step.setItemProcessor(new SimpleItemProcessor(new AbstractItemReader() {
step.setItemProcessor(new SimpleItemHandler(new AbstractItemReader() {
public Object read() throws Exception {
// do something non-trivial (and not Thread.sleep())
double foo = 1;