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

OPEN - issue BATCH-789: Remove mark/reset from ItemReader

Browse Source 
Shift chunk iteration into StepHandler - had to sacrifice interrupted check inside chunk
This commit is contained in:
dsyer
2008-08-20 15:24:05 +00:00
parent dbc24626f3
commit ef4f2900cc
19 changed files with 524 additions and 637 deletions
Download Patch File Download Diff File Expand all files Collapse all files

324
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/AbstractStepFactoryBean.java
View File

@@ -1,324 +0,0 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.step.item;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.validator.Validator;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.util.Assert;
/**
* Base class for factory beans for {@link StepHandlerStep}. Ensures that all
* the mandatory properties are set, and provides basic support for the
* {@link Step} interface responsibilities like start limit. Supports
* registration of {@link ItemStream}s and {@link StepListener}s.
*
* @see SimpleStepFactoryBean
* @see RepeatOperationsStepFactoryBean
*
* @author Dave Syer
*
*/
public abstract class AbstractStepFactoryBean<T,S> implements FactoryBean, BeanNameAware {
private String name;
private int startLimit = Integer.MAX_VALUE;
private boolean allowStartIfComplete;
private ItemReader<? extends T> itemReader;
private ItemWriter<? super S> itemWriter;
private PlatformTransactionManager transactionManager;
private TransactionAttribute transactionAttribute;
private JobRepository jobRepository;
private boolean singleton = true;
private Validator jobRepositoryValidator = new TransactionInterceptorValidator(1);
private ItemStream[] streams = new ItemStream[0];
private StepListener[] listeners = new StepListener[0];
private ItemProcessor<? super T, ? extends S> itemProcessor = new ItemProcessor<T, S>() {
@SuppressWarnings("unchecked")
public S process(T item) throws Exception {return (S)item;}
};
/**
*
*/
public AbstractStepFactoryBean() {
super();
}
/**
* Set the bean name property, which will become the name of the
* {@link Step} when it is created.
*
* @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String)
*/
public void setBeanName(String name) {
this.name = name;
}
/**
* Public getter for the String.
* @return the name
*/
public String getName() {
return name;
}
/**
* Public setter for the startLimit.
*
* @param startLimit the startLimit to set
*/
public void setStartLimit(int startLimit) {
this.startLimit = startLimit;
}
/**
* Public setter for the shouldAllowStartIfComplete.
*
* @param allowStartIfComplete the shouldAllowStartIfComplete to set
*/
public void setAllowStartIfComplete(boolean allowStartIfComplete) {
this.allowStartIfComplete = allowStartIfComplete;
}
/**
* @param itemReader the itemReader to set
*/
public void setItemReader(ItemReader<? extends T> itemReader) {
this.itemReader = itemReader;
}
/**
* @param itemWriter the itemWriter to set
*/
public void setItemWriter(ItemWriter<? super S> itemWriter) {
this.itemWriter = itemWriter;
}
/**
* @param itemProcessor the itemProcessor to set
*/
public void setItemProcessor(ItemProcessor<? super T, ? extends S> itemProcessor) {
this.itemProcessor = itemProcessor;
}
/**
* The streams to inject into the {@link Step}. Any instance of
* {@link ItemStream} can be used, and will then receive callbacks at the
* appropriate stage in the step.
*
* @param streams an array of listeners
*/
public void setStreams(ItemStream[] streams) {
this.streams = streams;
}
/**
* The listeners to inject into the {@link Step}. Any instance of
* {@link StepListener} can be used, and will then receive callbacks at the
* appropriate stage in the step.
*
* @param listeners an array of listeners
*/
public void setListeners(StepListener[] listeners) {
this.listeners = listeners;
}
/**
* Protected getter for the {@link StepListener}s.
* @return the listeners
*/
protected StepListener[] getListeners() {
return listeners;
}
/**
* Protected getter for the {@link ItemReader} for subclasses to use.
* @return the itemReader
*/
protected ItemReader<? extends T> getItemReader() {
return itemReader;
}
/**
* Protected getter for the {@link ItemWriter} for subclasses to use
* @return the itemWriter
*/
protected ItemWriter<? super S> getItemWriter() {
return itemWriter;
}
/**
* Protected getter for the {@link ItemProcessor} for subclasses to use
* @return the itemProcessor
*/
protected ItemProcessor<? super T, ? extends S> getItemProcessor() {
return itemProcessor;
}
/**
* Public setter for {@link JobRepository}.
*
* @param jobRepository is a mandatory dependence (no default).
*/
public void setJobRepository(JobRepository jobRepository) {
this.jobRepository = jobRepository;
}
/**
* Public setter for the {@link PlatformTransactionManager}.
*
* @param transactionManager the transaction manager to set
*/
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
/**
* Public setter for the {@link TransactionAttribute}.
* @param transactionAttribute the {@link TransactionAttribute} to set
*/
public void setTransactionAttribute(TransactionAttribute transactionAttribute) {
this.transactionAttribute = transactionAttribute;
}
/**
* Protected getter for the {@link TransactionAttribute} for subclasses
* only.
* @return the transactionAttribute
*/
protected TransactionAttribute getTransactionAttribute() {
return transactionAttribute!=null?transactionAttribute:new DefaultTransactionAttribute();
}
/**
* Create a {@link Step} from the configuration provided.
*
* @see org.springframework.beans.factory.FactoryBean#getObject()
*/
public final Object getObject() throws Exception {
StepHandlerStep step = new StepHandlerStep(getName());
applyConfiguration(step);
return step;
}
/**
* @param step
*
*/
protected void applyConfiguration(StepHandlerStep step) {
Assert.notNull(getItemReader(), "ItemReader must be provided");
Assert.notNull(getItemWriter(), "ItemWriter must be provided");
Assert.notNull(transactionManager, "TransactionManager must be provided");
jobRepositoryValidator.validate(jobRepository);
step.setTransactionManager(transactionManager);
if (transactionAttribute!=null) {
step.setTransactionAttribute(transactionAttribute);
}
step.setJobRepository(jobRepository);
step.setStartLimit(startLimit);
step.setAllowStartIfComplete(allowStartIfComplete);
step.setStreams(streams);
ItemReader<? extends T> itemReader = getItemReader();
ItemWriter<? super S> itemWriter = getItemWriter();
ItemProcessor<? super T, ? extends S> itemProcessor = getItemProcessor();
// Since we are going to wrap these things with listener callbacks we
// need to register them here because the step will not know we did
// that.
if (itemReader instanceof ItemStream) {
step.registerStream((ItemStream) itemReader);
}
if (itemReader instanceof StepExecutionListener) {
step.registerStepExecutionListener((StepExecutionListener) itemReader);
}
if (itemProcessor instanceof ItemStream) {
step.registerStream((ItemStream) itemProcessor);
}
if (itemProcessor instanceof StepExecutionListener) {
step.registerStepExecutionListener((StepExecutionListener) itemProcessor);
}
if (itemWriter instanceof ItemStream) {
step.registerStream((ItemStream) itemWriter);
}
if (itemWriter instanceof StepExecutionListener) {
step.registerStepExecutionListener((StepExecutionListener) itemWriter);
}
StepExecutionListener[] stepListeners = BatchListenerFactoryHelper.getStepListeners(listeners);
itemReader = BatchListenerFactoryHelper.getItemReader(itemReader, listeners);
itemWriter = BatchListenerFactoryHelper.getItemWriter(itemWriter, listeners);
// In case they are used by subclasses:
setItemReader(itemReader);
setItemWriter(itemWriter);
step.setStepExecutionListeners(stepListeners);
step.setItemHandler(new ItemOrientedStepHandler<T,S>(itemReader, itemProcessor, itemWriter));
}
public Class<Step> getObjectType() {
return Step.class;
}
/**
* Returns true by default, but in most cases a {@link Step} should not be
* treated as thread safe. Clients are recommended to create a new step for
* each job execution.
*
* @see org.springframework.beans.factory.FactoryBean#isSingleton()
*/
public boolean isSingleton() {
return this.singleton;
}
/**
* Public setter for the singleton flag.
* @param singleton the value to set. Defaults to true.
*/
public void setSingleton(boolean singleton) {
this.singleton = singleton;
}
}

50
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ItemOrientedStepHandler.java
View File

@@ -26,6 +26,10 @@ import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.MarkFailedException;
import org.springframework.batch.item.ResetFailedException;
import org.springframework.batch.repeat.ExitStatus;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.support.RepeatTemplate;
/**
* Simplest possible implementation of {@link StepHandler} with no skipping or
@@ -43,45 +47,57 @@ public class ItemOrientedStepHandler<T, S> implements StepHandler {
protected final Log logger = LogFactory.getLog(getClass());
private ItemReader<? extends T> itemReader;
private final ItemReader<? extends T> itemReader;
private ItemProcessor<? super T, ? extends S> itemProcessor;
private final ItemProcessor<? super T, ? extends S> itemProcessor;
private ItemWriter<? super S> itemWriter;
private final ItemWriter<? super S> itemWriter;
private final RepeatOperations repeatOperations;
/**
* @param itemReader
* @param itemProcessor
* @param itemWriter
* @param repeatOperations
*/
public ItemOrientedStepHandler(ItemReader<? extends T> itemReader,
ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter) {
ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter,
RepeatOperations repeatOperations) {
super();
this.itemReader = itemReader;
this.itemProcessor = itemProcessor;
this.itemWriter = itemWriter;
this.repeatOperations = repeatOperations;
}
/**
* Get the next item from {@link #read(StepContribution)} and if not null
* pass the item to {@link #write(Object, StepContribution)}. If the
* {@link ItemProcessor} returns null, the write is omitted and another
* item taken from the reader.
* {@link ItemProcessor} returns null, the write is omitted and another item
* taken from the reader.
*
* @see org.springframework.batch.core.step.item.StepHandler#handle(org.springframework.batch.core.StepContribution)
*/
public ExitStatus handle(StepContribution contribution) throws Exception {
boolean processed = false;
while (!processed) {
T item = read(contribution);
if (item == null) {
return ExitStatus.FINISHED;
public ExitStatus handle(final StepContribution contribution) throws Exception {
ExitStatus result = repeatOperations.iterate(new RepeatCallback() {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
boolean processed = false;
while (!processed) {
T item = read(contribution);
if (item == null) {
return ExitStatus.FINISHED;
}
// TODO: segregate read / write / filter count
contribution.incrementItemCount();
processed = write(item, contribution);
}
return ExitStatus.CONTINUABLE;
}
// TODO: segregate read / write / filter count
contribution.incrementItemCount();
processed = write(item, contribution);
}
return ExitStatus.CONTINUABLE;
});
return result;
}
/**

46
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBean.java
View File

@@ -17,57 +17,17 @@ package org.springframework.batch.core.step.item;
import org.springframework.batch.core.Step;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.support.RepeatTemplate;
/**
* Factory bean for {@link Step} implementations allowing registration of
* listeners and also direct injection of the {@link RepeatOperations} needed at
* step and chunk level.
*
* @deprecated use the {@link SimpleStepFactoryBean} instead
*
* @author Dave Syer
*
*/
public class RepeatOperationsStepFactoryBean<T,S> extends AbstractStepFactoryBean<T,S> {
private RepeatOperations chunkOperations = new RepeatTemplate();
private RepeatOperations stepOperations = new RepeatTemplate();
/**
* The {@link RepeatOperations} to use for the outer loop of the batch
* processing. Should be set up by the caller through a factory. Defaults to
* a plain {@link RepeatTemplate}.
*
* @param stepOperations a {@link RepeatOperations} instance.
*/
public void setStepOperations(RepeatOperations stepOperations) {
this.stepOperations = stepOperations;
}
/**
* The {@link RepeatOperations} to use for the inner loop of the batch
* processing. should be set up by the caller through a factory. defaults to
* a plain {@link RepeatTemplate}.
*
* @param chunkOperations a {@link RepeatOperations} instance.
*/
public void setChunkOperations(RepeatOperations chunkOperations) {
this.chunkOperations = chunkOperations;
}
/**
* @param step
*
*/
protected void applyConfiguration(StepHandlerStep step) {
super.applyConfiguration(step);
RepeatOperations chunkOperations = BatchListenerFactoryHelper.addChunkListeners(this.chunkOperations, getListeners());
step.setChunkOperations(chunkOperations);
step.setStepOperations(stepOperations);
}
public class RepeatOperationsStepFactoryBean<T,S> extends SimpleStepFactoryBean<T,S> {
}

363
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java
View File

@@ -18,17 +18,31 @@ package org.springframework.batch.core.step.item;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.validator.Validator;
import org.springframework.batch.repeat.CompletionPolicy;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.exception.DefaultExceptionHandler;
import org.springframework.batch.repeat.exception.ExceptionHandler;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.util.Assert;
/**
* Most common configuration options for simple steps should be found here. Use
* Most common configuration options for simple steps should be found here. Use
* this factory bean instead of creating a {@link Step} implementation manually.
*
* This factory does not support configuration of fault-tolerant behavior, use
@@ -39,21 +53,50 @@ import org.springframework.util.Assert;
* @author Dave Syer
*
*/
public class SimpleStepFactoryBean<T,S> extends AbstractStepFactoryBean<T,S> {
public class SimpleStepFactoryBean<T,S> implements FactoryBean, BeanNameAware {
private static final int DEFAULT_COMMIT_INTERVAL = 1;
private String name;
private int startLimit = Integer.MAX_VALUE;
private boolean allowStartIfComplete;
private ItemReader<? extends T> itemReader;
private ItemWriter<? super S> itemWriter;
private PlatformTransactionManager transactionManager;
private TransactionAttribute transactionAttribute;
private JobRepository jobRepository;
private boolean singleton = true;
private Validator jobRepositoryValidator = new TransactionInterceptorValidator(1);
private ItemStream[] streams = new ItemStream[0];
private StepListener[] listeners = new StepListener[0];
protected final Log logger = LogFactory.getLog(getClass());
private static final int DEFAULT_COMMIT_INTERVAL = 1;
private ItemProcessor<? super T, ? extends S> itemProcessor = new ItemProcessor<T, S>() {
@SuppressWarnings("unchecked")
public S process(T item) throws Exception {return (S)item;}
};
private int commitInterval = 0;
private TaskExecutor taskExecutor;
private StepHandler itemHandler;
private StepHandler stepHandler;
private RepeatTemplate stepOperations;
private RepeatOperations stepOperations;
private RepeatTemplate chunkOperations;
private RepeatOperations chunkOperations;
private ExceptionHandler exceptionHandler = new DefaultExceptionHandler();
@@ -61,6 +104,193 @@ public class SimpleStepFactoryBean<T,S> extends AbstractStepFactoryBean<T,S> {
private int throttleLimit = TaskExecutorRepeatTemplate.DEFAULT_THROTTLE_LIMIT;
/**
*
*/
public SimpleStepFactoryBean() {
super();
}
/**
* Set the bean name property, which will become the name of the
* {@link Step} when it is created.
*
* @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String)
*/
public void setBeanName(String name) {
this.name = name;
}
/**
* Public getter for the String.
* @return the name
*/
public String getName() {
return name;
}
/**
* Public setter for the startLimit.
*
* @param startLimit the startLimit to set
*/
public void setStartLimit(int startLimit) {
this.startLimit = startLimit;
}
/**
* Public setter for the shouldAllowStartIfComplete.
*
* @param allowStartIfComplete the shouldAllowStartIfComplete to set
*/
public void setAllowStartIfComplete(boolean allowStartIfComplete) {
this.allowStartIfComplete = allowStartIfComplete;
}
/**
* @param itemReader the itemReader to set
*/
public void setItemReader(ItemReader<? extends T> itemReader) {
this.itemReader = itemReader;
}
/**
* @param itemWriter the itemWriter to set
*/
public void setItemWriter(ItemWriter<? super S> itemWriter) {
this.itemWriter = itemWriter;
}
/**
* @param itemProcessor the itemProcessor to set
*/
public void setItemProcessor(ItemProcessor<? super T, ? extends S> itemProcessor) {
this.itemProcessor = itemProcessor;
}
/**
* The streams to inject into the {@link Step}. Any instance of
* {@link ItemStream} can be used, and will then receive callbacks at the
* appropriate stage in the step.
*
* @param streams an array of listeners
*/
public void setStreams(ItemStream[] streams) {
this.streams = streams;
}
/**
* The listeners to inject into the {@link Step}. Any instance of
* {@link StepListener} can be used, and will then receive callbacks at the
* appropriate stage in the step.
*
* @param listeners an array of listeners
*/
public void setListeners(StepListener[] listeners) {
this.listeners = listeners;
}
/**
* Protected getter for the {@link StepListener}s.
* @return the listeners
*/
protected StepListener[] getListeners() {
return listeners;
}
/**
* Protected getter for the {@link ItemReader} for subclasses to use.
* @return the itemReader
*/
protected ItemReader<? extends T> getItemReader() {
return itemReader;
}
/**
* Protected getter for the {@link ItemWriter} for subclasses to use
* @return the itemWriter
*/
protected ItemWriter<? super S> getItemWriter() {
return itemWriter;
}
/**
* Protected getter for the {@link ItemProcessor} for subclasses to use
* @return the itemProcessor
*/
protected ItemProcessor<? super T, ? extends S> getItemProcessor() {
return itemProcessor;
}
/**
* Public setter for {@link JobRepository}.
*
* @param jobRepository is a mandatory dependence (no default).
*/
public void setJobRepository(JobRepository jobRepository) {
this.jobRepository = jobRepository;
}
/**
* Public setter for the {@link PlatformTransactionManager}.
*
* @param transactionManager the transaction manager to set
*/
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
/**
* Public setter for the {@link TransactionAttribute}.
* @param transactionAttribute the {@link TransactionAttribute} to set
*/
public void setTransactionAttribute(TransactionAttribute transactionAttribute) {
this.transactionAttribute = transactionAttribute;
}
/**
* Protected getter for the {@link TransactionAttribute} for subclasses
* only.
* @return the transactionAttribute
*/
protected TransactionAttribute getTransactionAttribute() {
return transactionAttribute!=null?transactionAttribute:new DefaultTransactionAttribute();
}
/**
* Create a {@link Step} from the configuration provided.
*
* @see org.springframework.beans.factory.FactoryBean#getObject()
*/
public final Object getObject() throws Exception {
StepHandlerStep step = new StepHandlerStep(getName());
applyConfiguration(step);
return step;
}
public Class<Step> getObjectType() {
return Step.class;
}
/**
* Returns true by default, but in most cases a {@link Step} should not be
* treated as thread safe. Clients are recommended to create a new step for
* each job execution.
*
* @see org.springframework.beans.factory.FactoryBean#isSingleton()
*/
public boolean isSingleton() {
return this.singleton;
}
/**
* Public setter for the singleton flag.
* @param singleton the value to set. Defaults to true.
*/
public void setSingleton(boolean singleton) {
this.singleton = singleton;
}
/**
* Set the commit interval. Either set this or the chunkCompletionPolicy but
* not both.
@@ -88,16 +318,32 @@ public class SimpleStepFactoryBean<T,S> extends AbstractStepFactoryBean<T,S> {
* subclasses.
* @return the step operations
*/
protected RepeatTemplate getStepOperations() {
protected RepeatOperations getStepOperations() {
return stepOperations;
}
/**
* Public setter for the stepOperations.
* @param stepOperations the stepOperations to set
*/
public void setStepOperations(RepeatOperations stepOperations) {
this.stepOperations = stepOperations;
}
/**
* Public setter for the chunkOperations.
* @param chunkOperations the chunkOperations to set
*/
public void setChunkOperations(RepeatOperations chunkOperations) {
this.chunkOperations = chunkOperations;
}
/**
* Protected getter for the chunk operations to make them available in
* subclasses.
* @return the step operations
*/
protected RepeatTemplate getChunkOperations() {
protected RepeatOperations getChunkOperations() {
return chunkOperations;
}
@@ -139,19 +385,19 @@ public class SimpleStepFactoryBean<T,S> extends AbstractStepFactoryBean<T,S> {
}
/**
* Public getter for the ItemHandler.
* @return the ItemHandler
* Public getter for the {@link StepHandler}.
* @return the {@link StepHandler}
*/
protected StepHandler getItemHandler() {
return itemHandler;
protected StepHandler getStepHandler() {
return stepHandler;
}
/**
* Public setter for the ItemHandler.
* @param itemHandler the ItemHandler to set
* Public setter for the {@link StepHandler}.
* @param stepHandler the {@link StepHandler} to set
*/
protected void setItemHandler(StepHandler itemHandler) {
this.itemHandler = itemHandler;
protected void setStepHandler(StepHandler stepHandler) {
this.stepHandler = stepHandler;
}
/**
@@ -160,26 +406,83 @@ public class SimpleStepFactoryBean<T,S> extends AbstractStepFactoryBean<T,S> {
*/
protected void applyConfiguration(StepHandlerStep step) {
super.applyConfiguration(step);
Assert.notNull(getItemReader(), "ItemReader must be provided");
Assert.notNull(getItemWriter(), "ItemWriter must be provided");
Assert.notNull(transactionManager, "TransactionManager must be provided");
jobRepositoryValidator.validate(jobRepository);
chunkOperations = new RepeatTemplate();
chunkOperations.setCompletionPolicy(getChunkCompletionPolicy());
BatchListenerFactoryHelper.addChunkListeners(chunkOperations, getListeners());
step.setChunkOperations(chunkOperations);
step.setTransactionManager(transactionManager);
if (transactionAttribute!=null) {
step.setTransactionAttribute(transactionAttribute);
}
step.setJobRepository(jobRepository);
step.setStartLimit(startLimit);
step.setAllowStartIfComplete(allowStartIfComplete);
stepOperations = new RepeatTemplate();
step.setStreams(streams);
if (taskExecutor != null) {
TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate();
repeatTemplate.setTaskExecutor(taskExecutor);
repeatTemplate.setThrottleLimit(throttleLimit);
stepOperations = repeatTemplate;
ItemReader<? extends T> itemReader = getItemReader();
ItemWriter<? super S> itemWriter = getItemWriter();
ItemProcessor<? super T, ? extends S> itemProcessor = getItemProcessor();
// Since we are going to wrap these things with listener callbacks we
// need to register them here because the step will not know we did
// that.
if (itemReader instanceof ItemStream) {
step.registerStream((ItemStream) itemReader);
}
if (itemReader instanceof StepExecutionListener) {
step.registerStepExecutionListener((StepExecutionListener) itemReader);
}
if (itemProcessor instanceof ItemStream) {
step.registerStream((ItemStream) itemProcessor);
}
if (itemProcessor instanceof StepExecutionListener) {
step.registerStepExecutionListener((StepExecutionListener) itemProcessor);
}
if (itemWriter instanceof ItemStream) {
step.registerStream((ItemStream) itemWriter);
}
if (itemWriter instanceof StepExecutionListener) {
step.registerStepExecutionListener((StepExecutionListener) itemWriter);
}
stepOperations.setExceptionHandler(exceptionHandler);
StepExecutionListener[] stepListeners = BatchListenerFactoryHelper.getStepListeners(listeners);
itemReader = BatchListenerFactoryHelper.getItemReader(itemReader, listeners);
itemWriter = BatchListenerFactoryHelper.getItemWriter(itemWriter, listeners);
// In case they are used by subclasses:
setItemReader(itemReader);
setItemWriter(itemWriter);
step.setStepExecutionListeners(stepListeners);
if (chunkOperations == null) {
RepeatTemplate repeatTemplate = new RepeatTemplate();
repeatTemplate.setCompletionPolicy(getChunkCompletionPolicy());
chunkOperations = repeatTemplate;
}
BatchListenerFactoryHelper.addChunkListeners(chunkOperations, getListeners());
if (stepOperations == null) {
stepOperations = new RepeatTemplate();
if (taskExecutor != null) {
TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate();
repeatTemplate.setTaskExecutor(taskExecutor);
repeatTemplate.setThrottleLimit(throttleLimit);
stepOperations = repeatTemplate;
}
((RepeatTemplate) stepOperations).setExceptionHandler(exceptionHandler);
}
step.setStepOperations(stepOperations);
step.setItemHandler(new ItemOrientedStepHandler<T,S>(itemReader, itemProcessor, itemWriter, chunkOperations));
}
/**
@@ -190,7 +493,7 @@ public class SimpleStepFactoryBean<T,S> extends AbstractStepFactoryBean<T,S> {
Assert.state(!(chunkCompletionPolicy != null && commitInterval != 0),
"You must specify either a chunkCompletionPolicy or a commitInterval but not both.");
Assert.state(commitInterval >= 0, "The commitInterval must be positive or zero (for default value).");
if (chunkCompletionPolicy != null) {
return chunkCompletionPolicy;
}
@@ -201,4 +504,4 @@ public class SimpleStepFactoryBean<T,S> extends AbstractStepFactoryBean<T,S> {
return new SimpleCompletionPolicy(commitInterval);
}
}
}

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

@@ -16,6 +16,8 @@ import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.support.RepeatTemplate;
import org.springframework.batch.retry.RecoveryCallback;
import org.springframework.batch.retry.RetryCallback;
import org.springframework.batch.retry.RetryContext;
@@ -52,7 +54,7 @@ import org.springframework.batch.support.SubclassExceptionClassifier;
* @author Robert Kasanicky
*
*/
public class SkipLimitStepFactoryBean<T,S> extends SimpleStepFactoryBean<T,S> {
public class SkipLimitStepFactoryBean<T, S> extends SimpleStepFactoryBean<T, S> {
private int skipLimit = 0;
@@ -223,8 +225,11 @@ public class SkipLimitStepFactoryBean<T,S> extends SimpleStepFactoryBean<T,S> {
}
// Co-ordinate the retry policy with the exception handler:
getStepOperations().setExceptionHandler(
new SimpleRetryExceptionHandler(retryPolicy, getExceptionHandler(), fatalExceptionClasses));
RepeatOperations stepOperations = getStepOperations();
if (stepOperations instanceof RepeatTemplate) {
((RepeatTemplate) stepOperations).setExceptionHandler(new SimpleRetryExceptionHandler(retryPolicy,
getExceptionHandler(), fatalExceptionClasses));
}
RecoveryCallbackRetryPolicy recoveryCallbackRetryPolicy = new RecoveryCallbackRetryPolicy(retryPolicy) {
protected boolean recoverForException(Throwable ex) {
@@ -268,17 +273,14 @@ public class SkipLimitStepFactoryBean<T,S> extends SimpleStepFactoryBean<T,S> {
}
}
});
StatefulRetryStepHandler<T,S> itemHandler = new StatefulRetryStepHandler<T,S>(getItemReader(), getItemProcessor(), getItemWriter(),
retryTemplate, itemKeyGenerator, readSkipPolicy, writeSkipPolicy);
StatefulRetryStepHandler<T, S> itemHandler = new StatefulRetryStepHandler<T, S>(getItemReader(),
getItemProcessor(), getItemWriter(), getChunkOperations(), retryTemplate, itemKeyGenerator, readSkipPolicy,
writeSkipPolicy);
itemHandler.setSkipListeners(BatchListenerFactoryHelper.getSkipListeners(getListeners()));
step.setItemHandler(itemHandler);
}
else {
// This is the default in ItemOrientedStep anyway...
step.setItemHandler(new ItemOrientedStepHandler<T,S>(getItemReader(), getItemProcessor(), getItemWriter()));
}
}
@@ -306,7 +308,7 @@ public class SkipLimitStepFactoryBean<T,S> extends SimpleStepFactoryBean<T,S> {
* @author Dave Syer
*
*/
private static class StatefulRetryStepHandler<T,S> extends ItemOrientedStepHandler<T,S> {
private static class StatefulRetryStepHandler<T, S> extends ItemOrientedStepHandler<T, S> {
final private RetryOperations retryOperations;
@@ -324,10 +326,11 @@ public class SkipLimitStepFactoryBean<T,S> extends SimpleStepFactoryBean<T,S> {
* @param retryTemplate
* @param itemKeyGenerator
*/
public StatefulRetryStepHandler(ItemReader<? extends T> itemReader, ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter,
public StatefulRetryStepHandler(ItemReader<? extends T> itemReader,
ItemProcessor<? super T, ? extends S> itemProcessor, ItemWriter<? super S> itemWriter, RepeatOperations chunkOperations,
RetryOperations retryTemplate, ItemKeyGenerator itemKeyGenerator, ItemSkipPolicy readSkipPolicy,
ItemSkipPolicy writeSkipPolicy) {
super(itemReader, itemProcessor, itemWriter);
super(itemReader, itemProcessor, itemWriter, chunkOperations);
this.retryOperations = retryTemplate;
this.itemKeyGenerator = itemKeyGenerator;
this.readSkipPolicy = readSkipPolicy;
@@ -409,9 +412,6 @@ public class SkipLimitStepFactoryBean<T,S> extends SimpleStepFactoryBean<T,S> {
* {@link SkipListener} provided is called when retry attempts are
* exhausted. The listener callback (on write failure) will happen in
* the next transaction automatically.<br/>
*
* @see org.springframework.batch.core.step.item.SimpleStepHandler#write(java.lang.Object,
* org.springframework.batch.core.StepContribution)
*/
protected boolean write(final T item, final StepContribution contribution) throws Exception {
RecoveryRetryCallback retryCallback = new RecoveryRetryCallback(item, new RetryCallback() {

61
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepHandlerStep.java
View File

@@ -45,15 +45,9 @@ import org.springframework.transaction.interceptor.TransactionAttribute;
/**
* Simple implementation of executing the step as a set of chunks, each chunk
* surrounded by a transaction. The structure is therefore that of two nested
* loops, with transaction boundary around the whole inner loop. The outer loop
* is controlled by the step operations (
* {@link #setStepOperations(RepeatOperations)}), and the inner loop by the
* chunk operations ({@link #setChunkOperations(RepeatOperations)}). The inner
* loop should always be executed in a single thread, so the chunk operations
* should not do any concurrent execution. N.B. usually that means that the
* chunk operations should be a {@link RepeatTemplate} (which is the
* default).<br/>
* surrounded by a transaction. The structure is therefore that of a loop with
* transaction boundary inside the loop. The loop is controlled by the step
* operations ( {@link #setStepOperations(RepeatOperations)}).<br/>
*
* Clients can use interceptors in the step operations to intercept or listen to
* the iteration on a step-wide basis, for instance to get a callback when the
@@ -69,8 +63,6 @@ public class StepHandlerStep extends AbstractStep {
private static final Log logger = LogFactory.getLog(StepHandlerStep.class);
private RepeatOperations chunkOperations = new RepeatTemplate();
private RepeatOperations stepOperations = new RepeatTemplate();
// default to checking current thread for interruption.
@@ -173,17 +165,6 @@ public class StepHandlerStep extends AbstractStep {
this.stepOperations = stepOperations;
}
/**
* The {@link RepeatOperations} to use for the inner loop of the batch
* processing. should be set up by the caller through a factory. defaults to
* a plain {@link RepeatTemplate}.
*
* @param chunkOperations a {@link RepeatOperations} instance.
*/
public void setChunkOperations(RepeatOperations chunkOperations) {
this.chunkOperations = chunkOperations;
}
/**
* Setter for the {@link StepInterruptionPolicy}. The policy is used to
* check whether an external request has been made to interrupt the job
@@ -234,9 +215,6 @@ public class StepHandlerStep extends AbstractStep {
final StepContribution contribution = stepExecution.createStepContribution();
// Before starting a new transaction, check for
// interruption.
if (stepExecution.isTerminateOnly()) {
context.setTerminateOnly();
}
interruptionPolicy.checkInterrupted(stepExecution);
ExitStatus exitStatus = ExitStatus.CONTINUABLE;
@@ -248,7 +226,7 @@ public class StepHandlerStep extends AbstractStep {
try {
try {
exitStatus = processChunk(stepExecution, contribution);
exitStatus = itemHandler.handle(contribution);
}
catch (Error e) {
if (transactionAttribute.rollbackOn(e)) {
@@ -351,37 +329,6 @@ public class StepHandlerStep extends AbstractStep {
}
/**
* Execute a bunch of identical business logic operations all within a
* transaction. The transaction is programmatically started and stopped
* outside this method, so subclasses that override do not need to create a
* transaction.
* @param execution the current {@link StepExecution} which should be
* treated as read-only for the purposes of this method.
* @param contribution the current {@link StepContribution} which can accept
* changes to be aggregated later into the step execution.
*
* @return true if there is more data to process.
*/
protected ExitStatus processChunk(final StepExecution execution, final StepContribution contribution) {
ExitStatus result = chunkOperations.iterate(new RepeatCallback() {
public ExitStatus doInIteration(final RepeatContext context) throws Exception {
if (execution.isTerminateOnly()) {
context.setTerminateOnly();
}
// check for interruption before each item as well
interruptionPolicy.checkInterrupted(execution);
ExitStatus exitStatus = itemHandler.handle(contribution);
// check for interruption after each item as well
interruptionPolicy.checkInterrupted(execution);
return exitStatus;
}
});
return result;
}
/**
* @param stepExecution
* @param contribution

34
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ItemOrientedStepHandlerTests.java
View File

@@ -19,6 +19,7 @@ import static org.junit.Assert.assertEquals;
import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
@@ -32,6 +33,8 @@ import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.support.AbstractItemReader;
import org.springframework.batch.item.support.PassthroughItemProcessor;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
/**
* @author Dave Syer
@@ -42,22 +45,18 @@ public class ItemOrientedStepHandlerTests {
private StubItemReader itemReader = new StubItemReader();
private StubItemWriter itemWriter = new StubItemWriter();
private RepeatTemplate repeatTemplate = new RepeatTemplate();
@Before
public void setUp() {
repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(2));
}
@Test
public void testHandle() throws Exception {
ItemOrientedStepHandler<String, String> handler = new ItemOrientedStepHandler<String, String>(itemReader,
new PassthroughItemProcessor<String>(), itemWriter);
StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance(
123L, new JobParameters(), "job"))));
handler.handle(contribution);
assertEquals(1, itemReader.count);
assertEquals("1", itemWriter.values);
}
@Test
public void testHandleCompositeItem() throws Exception {
ItemOrientedStepHandler<String, String> handler = new ItemOrientedStepHandler<String, String>(itemReader,
new AgrgegateItemProcessor(), itemWriter);
new PassthroughItemProcessor<String>(), itemWriter, repeatTemplate);
StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance(
123L, new JobParameters(), "job"))));
handler.handle(contribution);
@@ -65,6 +64,17 @@ public class ItemOrientedStepHandlerTests {
assertEquals("12", itemWriter.values);
}
@Test
public void testHandleCompositeItem() throws Exception {
ItemOrientedStepHandler<String, String> handler = new ItemOrientedStepHandler<String, String>(itemReader,
new AgrgegateItemProcessor(), itemWriter, repeatTemplate);
StepContribution contribution = new StepContribution(new StepExecution("foo", new JobExecution(new JobInstance(
123L, new JobParameters(), "job"))));
handler.handle(contribution);
assertEquals(4, itemReader.count);
assertEquals("1234", itemWriter.values);
}
/**
* @author Dave Syer
*

2
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/RepeatOperationsStepFactoryBeanTests.java
View File

@@ -39,7 +39,7 @@ import org.springframework.batch.support.transaction.ResourcelessTransactionMana
*/
public class RepeatOperationsStepFactoryBeanTests extends TestCase {
private RepeatOperationsStepFactoryBean<String,String> factory = new RepeatOperationsStepFactoryBean<String,String>();
private SimpleStepFactoryBean<String,String> factory = new SimpleStepFactoryBean<String,String>();
private List<String> list;

67
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepFactoryBeanTests.java
View File

@@ -16,13 +16,18 @@
package org.springframework.batch.core.step.item;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import junit.framework.TestCase;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.JobExecution;
@@ -53,7 +58,7 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor;
/**
* Tests for {@link SimpleStepFactoryBean}.
*/
public class SimpleStepFactoryBeanTests extends TestCase {
public class SimpleStepFactoryBeanTests {
private List<Exception> recovered = new ArrayList<Exception>();
@@ -76,37 +81,15 @@ public class SimpleStepFactoryBeanTests extends TestCase {
}
};
protected void setUp() throws Exception {
super.setUp();
@Before
public void setUp() throws Exception {
job.setJobRepository(repository);
MapJobInstanceDao.clear();
MapJobExecutionDao.clear();
MapStepExecutionDao.clear();
}
private SimpleStepFactoryBean<String,String> getStepFactory(String arg) throws Exception {
return getStepFactory(new String[] { arg });
}
private SimpleStepFactoryBean<String,String> getStepFactory(String arg0, String arg1) throws Exception {
return getStepFactory(new String[] { arg0, arg1 });
}
private SimpleStepFactoryBean<String,String> getStepFactory(String[] args) throws Exception {
SimpleStepFactoryBean<String,String> factory = new SimpleStepFactoryBean<String,String>();
List<String> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(args));
reader = new ListItemReader<String>(items);
factory.setItemReader(reader);
factory.setItemWriter(writer);
factory.setJobRepository(repository);
factory.setTransactionManager(new ResourcelessTransactionManager());
factory.setBeanName("stepName");
return factory;
}
@Test
public void testSimpleJob() throws Exception {
job.setSteps(new ArrayList<Step>());
@@ -125,6 +108,7 @@ public class SimpleStepFactoryBeanTests extends TestCase {
assertTrue(written.contains("foo"));
}
@Test
public void testSimpleConcurrentJob() throws Exception {
job.setSteps(new ArrayList<Step>());
@@ -144,12 +128,13 @@ public class SimpleStepFactoryBeanTests extends TestCase {
assertTrue(written.contains("foo"));
}
@Test
public void testSimpleJobWithItemListeners() throws Exception {
final List<Throwable> throwables = new ArrayList<Throwable>();
RepeatTemplate chunkOperations = new RepeatTemplate();
// Always handle the exception a check it is the right one...
// Always handle the exception to check it is the right one...
chunkOperations.setExceptionHandler(new ExceptionHandler() {
public void handleException(RepeatContext context, Throwable throwable) throws RuntimeException {
throwables.add(throwable);
@@ -179,8 +164,8 @@ public class SimpleStepFactoryBeanTests extends TestCase {
}
} });
factory.setChunkOperations(chunkOperations);
StepHandlerStep step = (StepHandlerStep) factory.getObject();
step.setChunkOperations(chunkOperations);
job.setSteps(Collections.singletonList((Step) step));
@@ -194,6 +179,7 @@ public class SimpleStepFactoryBeanTests extends TestCase {
assertEquals(3, recovered.size());
}
@Test
public void testExceptionTerminates() throws Exception {
SimpleStepFactoryBean<String,String> factory = getStepFactory(new String[] { "foo", "bar", "spam" });
factory.setBeanName("exceptionStep");
@@ -217,6 +203,7 @@ public class SimpleStepFactoryBeanTests extends TestCase {
assertEquals(BatchStatus.FAILED, jobExecution.getStatus());
}
@Test
public void testExceptionHandler() throws Exception {
SimpleStepFactoryBean<String,String> factory = getStepFactory(new String[] { "foo", "bar", "spam" });
factory.setBeanName("exceptionStep");
@@ -240,6 +227,7 @@ public class SimpleStepFactoryBeanTests extends TestCase {
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
}
@Test
public void testChunkListeners() throws Exception {
String[] items = new String[] { "1", "2", "3", "4", "5", "6", "7" };
int commitInterval = 3;
@@ -282,11 +270,13 @@ public class SimpleStepFactoryBeanTests extends TestCase {
* Commit interval specified is not allowed to be zero or negative.
* @throws Exception
*/
@Test
public void testCommitIntervalMustBeGreaterThanZero() throws Exception {
SimpleStepFactoryBean<String,String> factory = getStepFactory("foo");
// nothing wrong here
factory.getObject();
factory = getStepFactory("foo");
// but exception expected after setting commit interval to value < 0
factory.setCommitInterval(-1);
try {
@@ -302,6 +292,7 @@ public class SimpleStepFactoryBeanTests extends TestCase {
* Commit interval specified is not allowed to be zero or negative.
* @throws Exception
*/
@Test
public void testCommitIntervalAndCompletionPolicyBothSet() throws Exception {
SimpleStepFactoryBean<String,String> factory = getStepFactory("foo");
@@ -318,4 +309,20 @@ public class SimpleStepFactoryBeanTests extends TestCase {
}
}
private SimpleStepFactoryBean<String,String> getStepFactory(String... args) throws Exception {
SimpleStepFactoryBean<String,String> factory = new SimpleStepFactoryBean<String,String>();
List<String> items = TransactionAwareProxyFactory.createTransactionalList();
items.addAll(Arrays.asList(args));
reader = new ListItemReader<String>(items);
factory.setItemReader(reader);
factory.setItemWriter(writer);
factory.setJobRepository(repository);
factory.setTransactionManager(new ResourcelessTransactionManager());
factory.setBeanName("stepName");
return factory;
}
}

23
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepHandler.java → spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleStepHandler.java
View File

@@ -18,6 +18,9 @@ package org.springframework.batch.core.step.item;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.PassthroughItemProcessor;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.RepeatTemplate;
/**
* Simplest possible implementation of {@link StepHandler} with no skipping or
@@ -28,12 +31,30 @@ import org.springframework.batch.item.support.PassthroughItemProcessor;
*/
public class SimpleStepHandler<T> extends ItemOrientedStepHandler<T, T> {
/**
*
*/
private static final RepeatTemplate repeatTemplate = new RepeatTemplate();
static {
// It's only for testing, and we don't want any infinite loops...
repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(6));
}
/**
* Creates a {@link PassthroughItemProcessor} and uses it to create an
* instance of {@link ItemOrientedStepHandler}.
*/
public SimpleStepHandler(ItemReader<T> itemReader, ItemWriter<T> itemWriter) {
super(itemReader, new PassthroughItemProcessor<T>(), itemWriter);
super(itemReader, new PassthroughItemProcessor<T>(), itemWriter, repeatTemplate);
}
/**
* Creates a {@link PassthroughItemProcessor} and uses it to create an
* instance of {@link ItemOrientedStepHandler}.
*/
public SimpleStepHandler(ItemReader<T> itemReader, ItemWriter<T> itemWriter, RepeatOperations repeatOperations) {
super(itemReader, new PassthroughItemProcessor<T>(), itemWriter, repeatOperations);
}
}

2
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StatefulRetryStepFactoryBeanTests.java
View File

@@ -136,7 +136,7 @@ public class StatefulRetryStepFactoryBeanTests extends TestCase {
factory.setItemReader(provider);
factory.setRetryLimit(10);
factory.setSkippableExceptionClasses(new Class[0]);
AbstractStep step = (AbstractStep) factory.getObject();
Step step = (Step) factory.getObject();
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
step.execute(stepExecution);

54
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepExecutorInterruptionTests.java
View File

@@ -68,18 +68,33 @@ public class StepExecutorInterruptionTests extends TestCase {
public void write(List<? extends Object> item) throws Exception {
}
};
step.setItemHandler(new SimpleStepHandler<Object>(new AbstractItemReader<Object>() {
public Object read() throws Exception {
return null;
}
}, itemWriter));
stepExecution = new StepExecution(step.getName(), jobExecution);
}
public void testInterruptChunk() throws Exception {
public void testInterruptStep() throws Exception {
Thread processingThread = createThread(stepExecution);
RepeatTemplate template = new RepeatTemplate();
// N.B, If we don't set the completion policy it might run forever
template.setCompletionPolicy(new SimpleCompletionPolicy(2));
step.setItemHandler(new SimpleStepHandler<Object>(new AbstractItemReader<Object>() {
public Object read() throws Exception {
// do something non-trivial (and not Thread.sleep())
double foo = 1;
for (int i = 2; i < 250; i++) {
foo = foo * i;
}
if (foo != 1) {
return new Double(foo);
}
else {
return null;
}
}
}, itemWriter, template));
processingThread.start();
Thread.sleep(100);
processingThread.interrupt();
@@ -96,14 +111,6 @@ public class StepExecutorInterruptionTests extends TestCase {
}
public void testInterruptStep() throws Exception {
RepeatTemplate template = new RepeatTemplate();
// N.B, If we don't set the completion policy it might run forever
template.setCompletionPolicy(new SimpleCompletionPolicy(2));
step.setChunkOperations(template);
testInterruptChunk();
}
public void testInterruptOnInterruptedException() throws Exception {
Thread processingThread = createThread(stepExecution);
@@ -176,23 +183,6 @@ public class StepExecutorInterruptionTests extends TestCase {
* @return
*/
private Thread createThread(final StepExecution stepExecution) {
step.setItemHandler(new SimpleStepHandler<Object>(new AbstractItemReader<Object>() {
public Object read() throws Exception {
// do something non-trivial (and not Thread.sleep())
double foo = 1;
for (int i = 2; i < 250; i++) {
foo = foo * i;
}
if (foo != 1) {
return new Double(foo);
}
else {
return null;
}
}
}, itemWriter));
Thread processingThread = new Thread() {
public void run() {
try {

12
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepHandlerStepIntegrationTests.java
View File

@@ -72,6 +72,8 @@ public class StepHandlerStepIntegrationTests {
private JobRepository jobRepository;
private RepeatTemplate chunkOperations;
private ItemReader<String> getReader(String[] args) {
return new ListItemReader<String>(Arrays.asList(args));
}
@@ -88,19 +90,17 @@ public class StepHandlerStepIntegrationTests {
jobRepositoryFactoryBean.setTransactionManager(transactionManager);
jobRepositoryFactoryBean.afterPropertiesSet();
jobRepository = (JobRepository) jobRepositoryFactoryBean.getObject();
RepeatTemplate template;
step = new StepHandlerStep("stepName");
step.setJobRepository(jobRepository);
step.setTransactionManager(transactionManager);
template = new RepeatTemplate();
RepeatTemplate template = new RepeatTemplate();
template.setCompletionPolicy(new SimpleCompletionPolicy(1));
step.setStepOperations(template);
// Only process one item:
template = new RepeatTemplate();
template.setCompletionPolicy(new SimpleCompletionPolicy(1));
step.setChunkOperations(template);
chunkOperations = new RepeatTemplate();
chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1));
job = new JobSupport("FOO");
@@ -121,7 +121,7 @@ public class StepHandlerStepIntegrationTests {
}
});
}
}));
}, chunkOperations));
JobExecution jobExecution = jobRepository.createJobExecution(job, new JobParameters());
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);

39
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepHandlerStepTests.java
View File

@@ -29,7 +29,6 @@ import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.UnexpectedJobExecutionException;
@@ -41,7 +40,6 @@ import org.springframework.batch.core.repository.dao.MapJobExecutionDao;
import org.springframework.batch.core.repository.dao.MapJobInstanceDao;
import org.springframework.batch.core.repository.dao.MapStepExecutionDao;
import org.springframework.batch.core.repository.support.SimpleJobRepository;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.core.step.JobRepositorySupport;
import org.springframework.batch.core.step.StepInterruptionPolicy;
import org.springframework.batch.item.ExecutionContext;
@@ -93,9 +91,12 @@ public class StepHandlerStepTests extends TestCase {
return new ListItemReader<String>(Arrays.asList(args));
}
private AbstractStep getStep(String[] strings) throws Exception {
private StepHandlerStep getStep(String[] strings) throws Exception {
StepHandlerStep step = new StepHandlerStep("stepName");
step.setItemHandler(new SimpleStepHandler<String>(getReader(strings), itemWriter));
// Only process one item:
RepeatTemplate template = new RepeatTemplate();
template.setCompletionPolicy(new SimpleCompletionPolicy(1));
step.setItemHandler(new SimpleStepHandler<String>(getReader(strings), itemWriter, template));
step.setJobRepository(new JobRepositorySupport());
step.setTransactionManager(transactionManager);
return step;
@@ -108,16 +109,11 @@ public class StepHandlerStepTests extends TestCase {
transactionManager = new ResourcelessTransactionManager();
RepeatTemplate template;
RepeatTemplate template = new RepeatTemplate();
template.setCompletionPolicy(new SimpleCompletionPolicy(1));
step = (StepHandlerStep) getStep(new String[] { "foo", "bar", "spam" });
template = new RepeatTemplate();
template.setCompletionPolicy(new SimpleCompletionPolicy(1));
step = getStep(new String[] { "foo", "bar", "spam" });
step.setStepOperations(template);
// Only process one item:
template = new RepeatTemplate();
template.setCompletionPolicy(new SimpleCompletionPolicy(1));
step.setChunkOperations(template);
job = new JobSupport("FOO");
jobInstance = new JobInstance(new Long(0), new JobParameters(), job.getName());
@@ -181,25 +177,6 @@ public class StepHandlerStepTests extends TestCase {
}
public void testChunkExecutor() throws Exception {
RepeatTemplate template = new RepeatTemplate();
// Only process one item:
template.setCompletionPolicy(new SimpleCompletionPolicy(1));
step.setChunkOperations(template);
JobExecution jobExecution = new JobExecution(jobInstance);
StepExecution stepExecution = new StepExecution(step.getName(), jobExecution);
StepContribution contribution = stepExecution.createStepContribution();
step.processChunk(stepExecution, contribution);
assertEquals(1, processed.size());
assertEquals(0, stepExecution.getItemCount());
assertEquals(1, contribution.getItemCount());
}
public void testRepository() throws Exception {
SimpleJobRepository repository = new SimpleJobRepository(new MapJobInstanceDao(), new MapJobExecutionDao(),

17
spring-batch-integration/.settings/org.eclipse.jdt.core.prefs
View File

@@ -1,4 +1,4 @@
#Thu Jul 24 14:51:08 CEST 2008
#Wed Aug 20 13:59:46 BST 2008
eclipse.preferences.version=1
org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.5
@@ -7,12 +7,13 @@ org.eclipse.jdt.core.compiler.compliance=1.5
org.eclipse.jdt.core.compiler.debug.lineNumber=generate
org.eclipse.jdt.core.compiler.debug.localVariable=generate
org.eclipse.jdt.core.compiler.debug.sourceFile=generate
org.eclipse.jdt.core.compiler.doc.comment.support=enabled
org.eclipse.jdt.core.compiler.problem.annotationSuperInterface=warning
org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
org.eclipse.jdt.core.compiler.problem.autoboxing=ignore
org.eclipse.jdt.core.compiler.problem.deprecation=warning
org.eclipse.jdt.core.compiler.problem.deprecationInDeprecatedCode=disabled
org.eclipse.jdt.core.compiler.problem.deprecationWhenOverridingDeprecatedMethod=disabled
org.eclipse.jdt.core.compiler.problem.deprecationWhenOverridingDeprecatedMethod=enabled
org.eclipse.jdt.core.compiler.problem.discouragedReference=warning
org.eclipse.jdt.core.compiler.problem.emptyStatement=ignore
org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
@@ -26,9 +27,21 @@ org.eclipse.jdt.core.compiler.problem.hiddenCatchBlock=warning
org.eclipse.jdt.core.compiler.problem.incompatibleNonInheritedInterfaceMethod=warning
org.eclipse.jdt.core.compiler.problem.incompleteEnumSwitch=ignore
org.eclipse.jdt.core.compiler.problem.indirectStaticAccess=ignore
org.eclipse.jdt.core.compiler.problem.invalidJavadoc=error
org.eclipse.jdt.core.compiler.problem.invalidJavadocTags=enabled
org.eclipse.jdt.core.compiler.problem.invalidJavadocTagsDeprecatedRef=enabled
org.eclipse.jdt.core.compiler.problem.invalidJavadocTagsNotVisibleRef=enabled
org.eclipse.jdt.core.compiler.problem.invalidJavadocTagsVisibility=default
org.eclipse.jdt.core.compiler.problem.localVariableHiding=ignore
org.eclipse.jdt.core.compiler.problem.methodWithConstructorName=warning
org.eclipse.jdt.core.compiler.problem.missingDeprecatedAnnotation=ignore
org.eclipse.jdt.core.compiler.problem.missingJavadocComments=ignore
org.eclipse.jdt.core.compiler.problem.missingJavadocCommentsOverriding=disabled
org.eclipse.jdt.core.compiler.problem.missingJavadocCommentsVisibility=public
org.eclipse.jdt.core.compiler.problem.missingJavadocTagDescription=return_tag
org.eclipse.jdt.core.compiler.problem.missingJavadocTags=ignore
org.eclipse.jdt.core.compiler.problem.missingJavadocTagsOverriding=disabled
org.eclipse.jdt.core.compiler.problem.missingJavadocTagsVisibility=public
org.eclipse.jdt.core.compiler.problem.missingOverrideAnnotation=ignore
org.eclipse.jdt.core.compiler.problem.missingSerialVersion=ignore
org.eclipse.jdt.core.compiler.problem.noEffectAssignment=warning

17
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java
View File

@@ -7,9 +7,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.listener.StepExecutionListenerSupport;
import org.springframework.batch.item.ClearFailedException;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.FlushFailedException;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemWriter;
@@ -78,20 +76,6 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
}
/**
* No-op.
* @see org.springframework.batch.item.ItemWriter#flush()
*/
public void flush() throws FlushFailedException {
}
/**
* No-op.
* @see org.springframework.batch.item.ItemWriter#clear()
*/
public void clear() throws ClearFailedException {
}
@Override
public void beforeStep(StepExecution stepExecution) {
localState.setStepExecution(stepExecution);
@@ -169,6 +153,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id ["
+ jobInstanceId + "] should have been [" + localState.getJobId() + "].");
localState.actual++;
// TODO: apply the skip count
ExitStatus result = payload.getExitStatus();
// TODO: check it can never be ExitStatus.FINISHED?
if (!result.isContinuable()) {

3
spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java
View File

@@ -53,7 +53,7 @@ public class ChunkMessageItemWriterIntegrationTests {
@Qualifier("replies")
private PollableChannel replies;
private SimpleStepFactoryBean<Object,Object> factory;
private SimpleStepFactoryBean<Object,Object> factory = new SimpleStepFactoryBean<Object,Object>();
private SimpleJobRepository jobRepository;
@@ -63,7 +63,6 @@ public class ChunkMessageItemWriterIntegrationTests {
@Before
public void setUp() {
factory = new SimpleStepFactoryBean<Object,Object>();
jobRepository = new SimpleJobRepository(new MapJobInstanceDao(),
new MapJobExecutionDao(), new MapStepExecutionDao(), new MapExecutionContextDao());
factory.setJobRepository(jobRepository);

12
spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterTests.java
View File

@@ -53,10 +53,6 @@ public class MessageChannelItemWriterTests {
assertEquals(Required.class, annotations[0].annotationType());
}
/**
* Test method for {@link org.springframework.batch.integration.item.MessageChannelItemWriter#write(java.lang.Object)}.
* @throws Exception
*/
@Test
public void testWrite() throws Exception {
DirectChannel channel = new DirectChannel();
@@ -70,10 +66,6 @@ public class MessageChannelItemWriterTests {
assertEquals("foo", message.getPayload());
}
/**
* Test method for {@link org.springframework.batch.integration.item.MessageChannelItemWriter#write(java.lang.Object)}.
* @throws Exception
*/
@Test
public void testWriteWithRollback() throws Exception {
DirectChannel channel = new DirectChannel();
@@ -93,10 +85,6 @@ public class MessageChannelItemWriterTests {
}
}
/**
* Test method for {@link org.springframework.batch.integration.item.MessageChannelItemWriter#write(java.lang.Object)}.
* @throws Exception
*/
@Test
public void testWriteWithRollbackOnEndpoint() throws Exception {
DirectChannel channel = new DirectChannel();

5
spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java
View File

@@ -39,7 +39,6 @@ import org.springframework.integration.channel.ThreadLocalChannel;
import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.ReflectionUtils;
@@ -85,10 +84,6 @@ public class MessageOrientedStepTests {
assertEquals(Required.class, annotations[0].annotationType());
}
/**
* Test method for
* {@link org.springframework.batch.integration.job.MessageOrientedStep#setSource(MessageSource)}.
*/
@Test
public void testSetReplyChannel() {
Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setSource",