OPEN - issue BATCH-429: Combine the core and execution modules
http://jira.springframework.org/browse/BATCH-429 Remove duplicate step factory beans (where did they come from?)
This commit is contained in:
dsyer
@@ -1,197 +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;
|
||||
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.beans.factory.BeanNameAware;
|
||||
import org.springframework.beans.factory.FactoryBean;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Base class for factory beans for {@link ItemOrientedStep}. Ensures that all
|
||||
* the mandatory properties are set, and provides basic support for the
|
||||
* {@link Step} interface responsibilities like start limit.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public abstract class AbstractStepFactoryBean implements FactoryBean, BeanNameAware {
|
||||
|
||||
private String name;
|
||||
|
||||
private int startLimit = Integer.MAX_VALUE;
|
||||
|
||||
private boolean allowStartIfComplete;
|
||||
|
||||
private ItemReader itemReader;
|
||||
|
||||
private ItemWriter itemWriter;
|
||||
|
||||
private PlatformTransactionManager transactionManager;
|
||||
|
||||
private JobRepository jobRepository;
|
||||
|
||||
private boolean singleton = true;
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
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 itemReader) {
|
||||
this.itemReader = itemReader;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param itemWriter the itemWriter to set
|
||||
*/
|
||||
public void setItemWriter(ItemWriter itemWriter) {
|
||||
this.itemWriter = itemWriter;
|
||||
}
|
||||
|
||||
/**
|
||||
* Protected getter for the {@link ItemReader} for subclasses to use.
|
||||
* @return the itemReader
|
||||
*/
|
||||
protected ItemReader getItemReader() {
|
||||
return itemReader;
|
||||
}
|
||||
|
||||
/**
|
||||
* Protected getter for the {@link ItemWriter} for subclasses to use
|
||||
* @return the itemWriter
|
||||
*/
|
||||
protected ItemWriter getItemWriter() {
|
||||
return itemWriter;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a {@link Step} from the configuration provided.
|
||||
*
|
||||
* @see org.springframework.beans.factory.FactoryBean#getObject()
|
||||
*/
|
||||
public final Object getObject() throws Exception {
|
||||
ItemOrientedStep step = new ItemOrientedStep(getName());
|
||||
applyConfiguration(step);
|
||||
return step;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param step
|
||||
*
|
||||
*/
|
||||
protected void applyConfiguration(ItemOrientedStep step) {
|
||||
|
||||
Assert.notNull(getItemReader(), "ItemReader must be provided");
|
||||
Assert.notNull(getItemWriter(), "ItemWriter must be provided");
|
||||
Assert.notNull(jobRepository, "JobRepository must be provided");
|
||||
Assert.notNull(transactionManager, "TransactionManager must be provided");
|
||||
|
||||
step.setItemHandler(new SimpleItemHandler(itemReader, itemWriter));
|
||||
step.setTransactionManager(transactionManager);
|
||||
step.setJobRepository(jobRepository);
|
||||
step.setStartLimit(startLimit);
|
||||
step.setAllowStartIfComplete(allowStartIfComplete);
|
||||
|
||||
}
|
||||
|
||||
public Class 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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,173 +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;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import org.springframework.batch.core.BatchListener;
|
||||
import org.springframework.batch.core.ChunkListener;
|
||||
import org.springframework.batch.core.ItemReadListener;
|
||||
import org.springframework.batch.core.ItemWriteListener;
|
||||
import org.springframework.batch.core.StepListener;
|
||||
import org.springframework.batch.core.listener.CompositeChunkListener;
|
||||
import org.springframework.batch.core.listener.CompositeItemReadListener;
|
||||
import org.springframework.batch.core.listener.CompositeItemWriteListener;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.batch.item.reader.DelegatingItemReader;
|
||||
import org.springframework.batch.item.writer.DelegatingItemWriter;
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
import org.springframework.batch.repeat.RepeatContext;
|
||||
import org.springframework.batch.repeat.RepeatOperations;
|
||||
import org.springframework.batch.repeat.listener.RepeatListenerSupport;
|
||||
import org.springframework.batch.repeat.support.RepeatTemplate;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* Package private helper for step factory beans.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
class BatchListenerFactoryHelper {
|
||||
|
||||
/**
|
||||
* @param itemReader2
|
||||
* @param listeners
|
||||
* @return
|
||||
*/
|
||||
public ItemReader getItemReader(ItemReader itemReader, BatchListener[] listeners) {
|
||||
|
||||
final CompositeItemReadListener multicaster = new CompositeItemReadListener();
|
||||
|
||||
for (int i = 0; i < listeners.length; i++) {
|
||||
BatchListener listener = listeners[i];
|
||||
if (listener instanceof ItemReadListener) {
|
||||
multicaster.register((ItemReadListener) listener);
|
||||
}
|
||||
}
|
||||
|
||||
itemReader = new DelegatingItemReader(itemReader) {
|
||||
public Object read() throws Exception {
|
||||
try {
|
||||
multicaster.beforeRead();
|
||||
Object item = super.read();
|
||||
multicaster.afterRead(item);
|
||||
return item;
|
||||
}
|
||||
catch (Exception e) {
|
||||
multicaster.onReadError(e);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
return itemReader;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param itemWriter2
|
||||
* @param listeners
|
||||
* @return
|
||||
*/
|
||||
public ItemWriter getItemWriter(ItemWriter itemWriter, BatchListener[] listeners) {
|
||||
final CompositeItemWriteListener multicaster = new CompositeItemWriteListener();
|
||||
|
||||
for (int i = 0; i < listeners.length; i++) {
|
||||
BatchListener listener = listeners[i];
|
||||
if (listener instanceof ItemWriteListener) {
|
||||
multicaster.register((ItemWriteListener) listener);
|
||||
}
|
||||
}
|
||||
|
||||
itemWriter = new DelegatingItemWriter(itemWriter) {
|
||||
public void write(Object item) throws Exception {
|
||||
try {
|
||||
multicaster.beforeWrite(item);
|
||||
super.write(item);
|
||||
multicaster.afterWrite();
|
||||
}
|
||||
catch (Exception e) {
|
||||
multicaster.onWriteError(e, item);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
return itemWriter;
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @param stepOperations
|
||||
* @param listeners
|
||||
* @return
|
||||
*/
|
||||
public RepeatOperations addChunkListeners(RepeatOperations stepOperations, BatchListener[] listeners) {
|
||||
|
||||
final CompositeChunkListener multicaster = new CompositeChunkListener();
|
||||
|
||||
boolean hasChunkListener = false;
|
||||
|
||||
for (int i = 0; i < listeners.length; i++) {
|
||||
BatchListener listener = listeners[i];
|
||||
if (listener instanceof ChunkListener) {
|
||||
hasChunkListener = true;
|
||||
}
|
||||
if (listener instanceof ChunkListener) {
|
||||
multicaster.register((ChunkListener) listener);
|
||||
}
|
||||
}
|
||||
|
||||
if (hasChunkListener) {
|
||||
|
||||
Assert.state(stepOperations instanceof RepeatTemplate,
|
||||
"Step operations is injected but not a RepeatTemplate, so chunk listeners cannot also be registered. "
|
||||
+ "Either inject a RepeatTemplate, or remove the ChunkListener.");
|
||||
|
||||
RepeatTemplate stepTemplate = (RepeatTemplate) stepOperations;
|
||||
stepTemplate.registerListener(new RepeatListenerSupport() {
|
||||
public void before(RepeatContext context) {
|
||||
multicaster.beforeChunk();
|
||||
}
|
||||
public void after(RepeatContext context, ExitStatus result) {
|
||||
multicaster.afterChunk();
|
||||
}
|
||||
});
|
||||
|
||||
}
|
||||
|
||||
return stepOperations;
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* @param listeners
|
||||
* @return
|
||||
*/
|
||||
public StepListener[] getStepListeners(BatchListener[] listeners) {
|
||||
List list = new ArrayList();
|
||||
for (int i = 0; i < listeners.length; i++) {
|
||||
BatchListener listener = listeners[i];
|
||||
if (listener instanceof StepListener) {
|
||||
list.add(listener);
|
||||
}
|
||||
}
|
||||
return (StepListener[]) list.toArray(new StepListener[list.size()]);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,246 +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;
|
||||
|
||||
import org.springframework.batch.core.BatchListener;
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.StepListener;
|
||||
import org.springframework.batch.core.step.support.LimitCheckingItemSkipPolicy;
|
||||
import org.springframework.batch.core.step.support.ListenerMulticaster;
|
||||
import org.springframework.batch.core.step.support.NeverSkipItemSkipPolicy;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.batch.repeat.exception.ExceptionHandler;
|
||||
import org.springframework.batch.repeat.exception.SimpleLimitExceptionHandler;
|
||||
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
|
||||
import org.springframework.batch.repeat.support.RepeatTemplate;
|
||||
import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate;
|
||||
import org.springframework.core.task.TaskExecutor;
|
||||
|
||||
/**
|
||||
* Most common configuration options for simple steps should be found here. Use
|
||||
* this factory bean instead of creating a {@link Step} implementation manually.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class DefaultStepFactoryBean extends AbstractStepFactoryBean {
|
||||
|
||||
private int skipLimit = 0;
|
||||
|
||||
private int commitInterval = 0;
|
||||
|
||||
private ItemStream[] streams = new ItemStream[0];
|
||||
|
||||
private BatchListener[] listeners = new BatchListener[0];
|
||||
|
||||
private ListenerMulticaster listener = new ListenerMulticaster();
|
||||
|
||||
private TaskExecutor taskExecutor;
|
||||
|
||||
private ItemHandler itemHandler;
|
||||
|
||||
private RepeatTemplate stepOperations;
|
||||
|
||||
private SimpleLimitExceptionHandler exceptionHandler;
|
||||
|
||||
/**
|
||||
* Set the commit interval.
|
||||
*
|
||||
* @param commitInterval
|
||||
*/
|
||||
public void setCommitInterval(int commitInterval) {
|
||||
this.commitInterval = commitInterval;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for a limit that determines skip policy. If this value is
|
||||
* positive then an exception in chunk processing will cause the item to be
|
||||
* skipped and no exception propagated until the limit is reached. If it is
|
||||
* zero then all exceptions will be propagated from the chunk and cause the
|
||||
* step to abort.
|
||||
*
|
||||
* @param skipLimit the value to set. Default is 0 (never skip).
|
||||
*/
|
||||
public void setSkipLimit(int skipLimit) {
|
||||
this.skipLimit = skipLimit;
|
||||
}
|
||||
|
||||
/**
|
||||
* The listeners to inject into the {@link Step}. Any instance of
|
||||
* {@link BatchListener} can be used, and will then receive callbacks at the
|
||||
* appropriate stage in the step.
|
||||
*
|
||||
* @param listeners an array of listeners
|
||||
*/
|
||||
public void setListeners(BatchListener[] listeners) {
|
||||
this.listeners = listeners;
|
||||
}
|
||||
|
||||
/**
|
||||
* Protected getter for the step operations to make them available in
|
||||
* subclasses.
|
||||
* @return the step operations
|
||||
*/
|
||||
protected RepeatTemplate getStepOperations() {
|
||||
return stepOperations;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the SimpleLimitExceptionHandler.
|
||||
* @param exceptionHandler the exceptionHandler to set
|
||||
*/
|
||||
public void setExceptionHandler(SimpleLimitExceptionHandler exceptionHandler) {
|
||||
this.exceptionHandler = exceptionHandler;
|
||||
}
|
||||
|
||||
/**
|
||||
* Protected getter for the {@link ExceptionHandler}.
|
||||
* @return the {@link ExceptionHandler}
|
||||
*/
|
||||
protected ExceptionHandler getExceptionHandler() {
|
||||
return exceptionHandler;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link TaskExecutor}. If this is set, then it will
|
||||
* be used to execute the chunk processing inside the {@link Step}.
|
||||
*
|
||||
* @param taskExecutor the taskExecutor to set
|
||||
*/
|
||||
public void setTaskExecutor(TaskExecutor taskExecutor) {
|
||||
this.taskExecutor = taskExecutor;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public getter for the ItemProcessor.
|
||||
* @return the itemProcessor
|
||||
*/
|
||||
protected ItemHandler getItemHandler() {
|
||||
return itemHandler;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the ItemProcessor.
|
||||
* @param itemHandler the itemProcessor to set
|
||||
*/
|
||||
protected void setItemHandler(ItemHandler itemHandler) {
|
||||
this.itemHandler = itemHandler;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param step
|
||||
*
|
||||
*/
|
||||
protected void applyConfiguration(ItemOrientedStep step) {
|
||||
|
||||
super.applyConfiguration(step);
|
||||
|
||||
step.setStreams(streams);
|
||||
|
||||
for (int i = 0; i < listeners.length; i++) {
|
||||
BatchListener listener = listeners[i];
|
||||
if (listener instanceof StepListener) {
|
||||
step.registerStepListener((StepListener) listener);
|
||||
}
|
||||
else {
|
||||
this.listener.register(listener);
|
||||
}
|
||||
}
|
||||
|
||||
ItemReader itemReader = getItemReader();
|
||||
ItemWriter itemWriter = getItemWriter();
|
||||
|
||||
// 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 StepListener) {
|
||||
step.registerStepListener((StepListener) itemReader);
|
||||
}
|
||||
if (itemWriter instanceof ItemStream) {
|
||||
step.registerStream((ItemStream) itemWriter);
|
||||
}
|
||||
if (itemWriter instanceof StepListener) {
|
||||
step.registerStepListener((StepListener) itemWriter);
|
||||
}
|
||||
|
||||
BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper();
|
||||
|
||||
if (commitInterval > 0) {
|
||||
RepeatTemplate chunkOperations = new RepeatTemplate();
|
||||
chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(commitInterval));
|
||||
helper.addChunkListeners(chunkOperations, listeners);
|
||||
step.setChunkOperations(chunkOperations);
|
||||
}
|
||||
|
||||
StepListener[] stepListeners = helper.getStepListeners(listeners);
|
||||
itemReader = helper.getItemReader(itemReader, listeners);
|
||||
itemWriter = helper.getItemWriter(itemWriter, listeners);
|
||||
stepOperations = new RepeatTemplate();
|
||||
|
||||
// In case they are used by subclasses:
|
||||
setItemReader(itemReader);
|
||||
setItemWriter(itemWriter);
|
||||
|
||||
step.setStepListeners(stepListeners);
|
||||
|
||||
if (taskExecutor != null) {
|
||||
TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate();
|
||||
repeatTemplate.setTaskExecutor(taskExecutor);
|
||||
stepOperations = repeatTemplate;
|
||||
}
|
||||
|
||||
step.setStepOperations(stepOperations);
|
||||
|
||||
ItemSkipPolicyItemHandler itemProcessor = new ItemSkipPolicyItemHandler(itemReader, itemWriter);
|
||||
|
||||
if (skipLimit > 0) {
|
||||
/*
|
||||
* If there is a skip limit (not the default) then we are prepared
|
||||
* to absorb exceptions at the step level because the failed items
|
||||
* will never re-appear after a rollback.
|
||||
*/
|
||||
itemProcessor.setItemSkipPolicy(new LimitCheckingItemSkipPolicy(skipLimit));
|
||||
setExceptionHandler(new SimpleLimitExceptionHandler(skipLimit));
|
||||
stepOperations.setExceptionHandler(getExceptionHandler());
|
||||
step.setStepOperations(stepOperations);
|
||||
}
|
||||
else {
|
||||
// This is the default in ItemOrientedStep anyway...
|
||||
itemProcessor.setItemSkipPolicy(new NeverSkipItemSkipPolicy());
|
||||
}
|
||||
|
||||
setItemHandler(itemProcessor);
|
||||
step.setItemHandler(itemProcessor);
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,103 +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;
|
||||
|
||||
import org.springframework.batch.core.ItemSkipPolicy;
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.step.support.NeverSkipItemSkipPolicy;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.batch.item.Skippable;
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
|
||||
/**
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class ItemSkipPolicyItemHandler extends SimpleItemHandler {
|
||||
|
||||
private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy();
|
||||
|
||||
/**
|
||||
* @param itemReader
|
||||
* @param itemWriter
|
||||
*/
|
||||
public ItemSkipPolicyItemHandler(ItemReader itemReader, ItemWriter itemWriter) {
|
||||
super(itemReader, itemWriter);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param itemSkipPolicy
|
||||
*/
|
||||
public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) {
|
||||
this.itemSkipPolicy = itemSkipPolicy;
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute the business logic, delegating to the reader and writer.
|
||||
* Subclasses could extend the behaviour as long as they always return the
|
||||
* value of this method call in their superclass.<br/>
|
||||
*
|
||||
* Read from the {@link ItemReader} and process (if not null) with the
|
||||
* {@link ItemWriter}.<br/>
|
||||
*
|
||||
* If there is an exception and the reader or writer implements
|
||||
* {@link Skippable} then the skip method is called.
|
||||
*
|
||||
* @param contribution the current step
|
||||
* @return {@link ExitStatus#CONTINUABLE} if there is more processing to do
|
||||
* @throws Exception if there is an error
|
||||
*/
|
||||
public ExitStatus handle(StepContribution contribution) throws Exception {
|
||||
ExitStatus exitStatus = ExitStatus.CONTINUABLE;
|
||||
|
||||
try {
|
||||
|
||||
exitStatus = super.handle(contribution);
|
||||
|
||||
}
|
||||
catch (Exception e) {
|
||||
|
||||
if (itemSkipPolicy.shouldSkip(e, contribution.getSkipCount())) {
|
||||
contribution.incrementSkipCount();
|
||||
skip();
|
||||
}
|
||||
else {
|
||||
// Rethrow so that outer transaction is rolled back properly
|
||||
throw e;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
return exitStatus;
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark the current item as skipped if possible. If the reader and / or
|
||||
* writer are {@link Skippable} then delegate to them in that order.
|
||||
*
|
||||
* @see org.springframework.batch.item.Skippable#skip()
|
||||
*/
|
||||
private void skip() {
|
||||
if (getItemReader() instanceof Skippable) {
|
||||
((Skippable) getItemReader()).skip();
|
||||
}
|
||||
if (getItemWriter() instanceof Skippable) {
|
||||
((Skippable) getItemWriter()).skip();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,139 +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;
|
||||
|
||||
import org.springframework.batch.core.BatchListener;
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.StepListener;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
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.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class RepeatOperationsStepFactoryBean extends AbstractStepFactoryBean {
|
||||
|
||||
private ItemStream[] streams = new ItemStream[0];
|
||||
|
||||
private BatchListener[] listeners = new BatchListener[0];
|
||||
|
||||
private RepeatOperations chunkOperations = new RepeatTemplate();
|
||||
|
||||
private RepeatOperations stepOperations = new RepeatTemplate();
|
||||
|
||||
/**
|
||||
* 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 BatchListener} can be used, and will then receive callbacks at the
|
||||
* appropriate stage in the step.
|
||||
*
|
||||
* @param listeners an array of listeners
|
||||
*/
|
||||
public void setListeners(BatchListener[] listeners) {
|
||||
this.listeners = listeners;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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(ItemOrientedStep step) {
|
||||
|
||||
super.applyConfiguration(step);
|
||||
|
||||
step.setStreams(streams);
|
||||
|
||||
ItemReader itemReader = getItemReader();
|
||||
ItemWriter itemWriter = getItemWriter();
|
||||
|
||||
/*
|
||||
* 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 StepListener) {
|
||||
step.registerStepListener((StepListener) itemReader);
|
||||
}
|
||||
if (itemWriter instanceof ItemStream) {
|
||||
step.registerStream((ItemStream) itemWriter);
|
||||
}
|
||||
if (itemWriter instanceof StepListener) {
|
||||
step.registerStepListener((StepListener) itemWriter);
|
||||
}
|
||||
|
||||
BatchListenerFactoryHelper helper = new BatchListenerFactoryHelper();
|
||||
|
||||
StepListener[] stepListeners = helper.getStepListeners(listeners);
|
||||
itemReader = helper.getItemReader(itemReader, listeners);
|
||||
itemWriter = helper.getItemWriter(itemWriter, listeners);
|
||||
RepeatOperations stepOperations = helper.addChunkListeners(this.stepOperations, listeners);
|
||||
|
||||
// In case they are used by subclasses:
|
||||
setItemReader(itemReader);
|
||||
setItemWriter(itemWriter);
|
||||
|
||||
step.setStepListeners(stepListeners);
|
||||
step.setItemHandler(new SimpleItemHandler(itemReader, itemWriter));
|
||||
|
||||
step.setChunkOperations(chunkOperations);
|
||||
step.setStepOperations(stepOperations);
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,114 +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;
|
||||
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.item.ClearFailedException;
|
||||
import org.springframework.batch.item.FlushFailedException;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.batch.item.MarkFailedException;
|
||||
import org.springframework.batch.item.ResetFailedException;
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
|
||||
/**
|
||||
* 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 SimpleItemHandler implements ItemHandler {
|
||||
|
||||
private ItemReader itemReader;
|
||||
|
||||
private ItemWriter itemWriter;
|
||||
|
||||
/**
|
||||
* @param itemReader
|
||||
* @param itemWriter
|
||||
*/
|
||||
public SimpleItemHandler(ItemReader itemReader, ItemWriter itemWriter) {
|
||||
super();
|
||||
this.itemReader = itemReader;
|
||||
this.itemWriter = itemWriter;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public getter for the ItemReader.
|
||||
* @return the itemReader
|
||||
*/
|
||||
public ItemReader getItemReader() {
|
||||
return itemReader;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public getter for the ItemWriter.
|
||||
* @return the itemWriter
|
||||
*/
|
||||
public ItemWriter getItemWriter() {
|
||||
return itemWriter;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read from the {@link ItemReader} and process (if not null) with the
|
||||
* {@link ItemWriter}.
|
||||
*
|
||||
* @see org.springframework.batch.core.step.ItemHandler#handle(org.springframework.batch.core.StepContribution)
|
||||
*/
|
||||
public ExitStatus handle(StepContribution contribution) throws Exception {
|
||||
Object item = itemReader.read();
|
||||
if (item == null) {
|
||||
return ExitStatus.FINISHED;
|
||||
}
|
||||
itemWriter.write(item);
|
||||
return ExitStatus.CONTINUABLE;
|
||||
}
|
||||
|
||||
/**
|
||||
* @throws MarkFailedException
|
||||
* @see org.springframework.batch.item.ItemReader#mark()
|
||||
*/
|
||||
public void mark() throws MarkFailedException {
|
||||
itemReader.mark();
|
||||
}
|
||||
|
||||
/**
|
||||
* @throws ResetFailedException
|
||||
* @see org.springframework.batch.item.ItemReader#reset()
|
||||
*/
|
||||
public void reset() throws ResetFailedException {
|
||||
itemReader.reset();
|
||||
}
|
||||
|
||||
/**
|
||||
* @throws ClearFailedException
|
||||
* @see org.springframework.batch.item.ItemWriter#clear()
|
||||
*/
|
||||
public void clear() throws ClearFailedException {
|
||||
itemWriter.clear();
|
||||
}
|
||||
|
||||
/**
|
||||
* @throws FlushFailedException
|
||||
* @see org.springframework.batch.item.ItemWriter#flush()
|
||||
*/
|
||||
public void flush() throws FlushFailedException {
|
||||
itemWriter.flush();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,211 +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;
|
||||
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.step.support.SimpleRetryExceptionHandler;
|
||||
import org.springframework.batch.item.ItemKeyGenerator;
|
||||
import org.springframework.batch.item.ItemReader;
|
||||
import org.springframework.batch.item.ItemRecoverer;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.batch.repeat.ExitStatus;
|
||||
import org.springframework.batch.retry.RetryListener;
|
||||
import org.springframework.batch.retry.RetryOperations;
|
||||
import org.springframework.batch.retry.RetryPolicy;
|
||||
import org.springframework.batch.retry.backoff.BackOffPolicy;
|
||||
import org.springframework.batch.retry.callback.ItemReaderRetryCallback;
|
||||
import org.springframework.batch.retry.policy.ItemReaderRetryPolicy;
|
||||
import org.springframework.batch.retry.policy.SimpleRetryPolicy;
|
||||
import org.springframework.batch.retry.support.RetryTemplate;
|
||||
|
||||
/**
|
||||
* Factory bean for step that executes its item processing with a stateful
|
||||
* retry. Failed items are never skipped, but always cause a rollback. Before a
|
||||
* rollback, the {@link Step} makes a record of the failed item, caching it
|
||||
* under a key given by the {@link ItemKeyGenerator}. Then when it is
|
||||
* re-presented by the {@link ItemReader} it is recognised and retried up to a
|
||||
* limit given by the {@link RetryPolicy}. When the retry is exhausted instead
|
||||
* of the item being skipped it is handled by an {@link ItemRecoverer}.<br/>
|
||||
*
|
||||
* The skipLimit property is still used to control the overall exception
|
||||
* handling policy. Only exhausted retries count against the exception handler,
|
||||
* instead of counting all exceptions.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class StatefulRetryStepFactoryBean extends DefaultStepFactoryBean {
|
||||
|
||||
private ItemKeyGenerator itemKeyGenerator;
|
||||
|
||||
private ItemRecoverer itemRecoverer;
|
||||
|
||||
private int retryLimit;
|
||||
|
||||
private Class[] retryableExceptionClasses;
|
||||
|
||||
private BackOffPolicy backOffPolicy;
|
||||
|
||||
private RetryListener[] retryListeners;
|
||||
|
||||
/**
|
||||
* Public setter for the retry limit. Each item can be retried up to this
|
||||
* limit.
|
||||
* @param retryLimit the retry limit to set
|
||||
*/
|
||||
public void setRetryLimit(int retryLimit) {
|
||||
this.retryLimit = retryLimit;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the Class[].
|
||||
* @param retryableExceptionClasses the retryableExceptionClasses to set
|
||||
*/
|
||||
public void setRetryableExceptionClasses(Class[] retryableExceptionClasses) {
|
||||
this.retryableExceptionClasses = retryableExceptionClasses;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link BackOffPolicy}.
|
||||
* @param backOffPolicy the {@link BackOffPolicy} to set
|
||||
*/
|
||||
public void setBackOffPolicy(BackOffPolicy backOffPolicy) {
|
||||
this.backOffPolicy = backOffPolicy;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link RetryListener}s.
|
||||
* @param retryListeners the {@link RetryListener}s to set
|
||||
*/
|
||||
public void setRetryListeners(RetryListener[] retryListeners) {
|
||||
this.retryListeners = retryListeners;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link ItemKeyGenerator} which will be used to
|
||||
* cache failed items between transactions. If it is not injected but the
|
||||
* reader or writer implement {@link ItemKeyGenerator}, one of those will
|
||||
* be used instead (preferring the reader to the writer if both would be
|
||||
* appropriate). If neither can be used, then the default will be to just
|
||||
* use the item itself as a cache key.
|
||||
*
|
||||
* @param itemKeyGenerator the {@link ItemKeyGenerator} to set
|
||||
*/
|
||||
public void setItemKeyGenerator(ItemKeyGenerator itemKeyGenerator) {
|
||||
this.itemKeyGenerator = itemKeyGenerator;
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link ItemRecoverer}. If this is set the
|
||||
* {@link ItemRecoverer#recover(Object, Throwable)} will be called when
|
||||
* retry is exhausted, and within the business transaction (which will not
|
||||
* roll back because of any other item-related errors).
|
||||
*
|
||||
* @param itemRecoverer the {@link ItemRecoverer} to set
|
||||
*/
|
||||
public void setItemRecoverer(ItemRecoverer itemRecoverer) {
|
||||
this.itemRecoverer = itemRecoverer;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param step
|
||||
*
|
||||
*/
|
||||
protected void applyConfiguration(ItemOrientedStep step) {
|
||||
|
||||
super.applyConfiguration(step);
|
||||
|
||||
if (retryLimit > 0) {
|
||||
|
||||
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy(retryLimit);
|
||||
if (retryableExceptionClasses != null) {
|
||||
retryPolicy.setRetryableExceptionClasses(retryableExceptionClasses);
|
||||
}
|
||||
|
||||
// Co-ordinate the retry policy with the exception handler:
|
||||
getStepOperations()
|
||||
.setExceptionHandler(new SimpleRetryExceptionHandler(retryPolicy, getExceptionHandler()));
|
||||
|
||||
ItemReaderRetryCallback retryCallback = new ItemReaderRetryCallback(getItemReader(), itemKeyGenerator,
|
||||
getItemWriter());
|
||||
retryCallback.setRecoverer(itemRecoverer);
|
||||
ItemReaderRetryPolicy itemProviderRetryPolicy = new ItemReaderRetryPolicy(retryPolicy);
|
||||
|
||||
RetryTemplate retryTemplate = new RetryTemplate();
|
||||
if (retryListeners!=null) {
|
||||
retryTemplate.setListeners(retryListeners);
|
||||
}
|
||||
retryTemplate.setRetryPolicy(itemProviderRetryPolicy);
|
||||
if (backOffPolicy != null) {
|
||||
retryTemplate.setBackOffPolicy(backOffPolicy);
|
||||
}
|
||||
|
||||
StatefulRetryItemHandler itemProcessor = new StatefulRetryItemHandler(getItemReader(), getItemWriter(),
|
||||
retryTemplate, retryCallback);
|
||||
|
||||
step.setItemHandler(itemProcessor);
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static class StatefulRetryItemHandler extends SimpleItemHandler {
|
||||
|
||||
final private RetryOperations retryOperations;
|
||||
|
||||
final private ItemReaderRetryCallback retryCallback;
|
||||
|
||||
/**
|
||||
* @param itemReader
|
||||
* @param itemWriter
|
||||
* @param retryCallback
|
||||
* @param retryTemplate
|
||||
*/
|
||||
public StatefulRetryItemHandler(ItemReader itemReader, ItemWriter itemWriter, RetryOperations retryTemplate,
|
||||
ItemReaderRetryCallback retryCallback) {
|
||||
super(itemReader, itemWriter);
|
||||
this.retryOperations = retryTemplate;
|
||||
this.retryCallback = retryCallback;
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute the business logic, delegating to the reader and writer.
|
||||
* Subclasses could extend the behaviour as long as they always return
|
||||
* the value of this method call in their superclass.<br/>
|
||||
*
|
||||
* Read from the {@link ItemReader} and process (if not null) with the
|
||||
* {@link ItemWriter}. The call to {@link ItemWriter} is wrapped in a
|
||||
* stateful retry. In that case the {@link ItemRecoverer} is used (if
|
||||
* provided) in the case of an exception to apply alternate processing
|
||||
* to the item. If the stateful retry is in place then the recovery will
|
||||
* happen in the next transaction automatically, otherwise it might be
|
||||
* necessary for clients to make the recover method transactional with
|
||||
* appropriate propagation behaviour (probably REQUIRES_NEW because the
|
||||
* call will happen in the context of a transaction that is about to
|
||||
* rollback).<br/>
|
||||
*
|
||||
* @param contribution the current step
|
||||
* @return {@link ExitStatus#CONTINUABLE} if there is more processing to
|
||||
* do
|
||||
* @throws Exception if there is an error
|
||||
*/
|
||||
public ExitStatus handle(StepContribution contribution) throws Exception {
|
||||
return new ExitStatus(retryOperations.execute(retryCallback) != null);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user