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

RESOLVED - issue BATCH-1458: Add FlowStep: a Step implementation that executes a flow

Browse Source
This commit is contained in:
dsyer
2009-12-08 17:37:56 +00:00
parent 9f3c61c650
commit 1ed2e13b3b
21 changed files with 995 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java
View File

@@ -23,6 +23,7 @@ import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.parsing.CompositeComponentDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.GenericBeanDefinition;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.xml.ParserContext;
@@ -57,6 +58,8 @@ public abstract class AbstractStepParser {
private static final String TASKLET_ELE = "tasklet";
private static final String FLOW_ELE = "flow";
private static final String CHUNK_ELE = "chunk";
private static final String LISTENERS_ELE = "listeners";
@@ -79,18 +82,19 @@ public abstract class AbstractStepParser {
*/
protected AbstractBeanDefinition parseStep(Element stepElement, ParserContext parserContext, String jobFactoryRef) {
AbstractBeanDefinition bd = new GenericBeanDefinition();
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition();
AbstractBeanDefinition bd = builder.getRawBeanDefinition();
@SuppressWarnings("unchecked")
List<Element> taskletElements = DomUtils.getChildElementsByTagName(stepElement, TASKLET_ELE);
if (taskletElements.size() == 1) {
Element taskletElement = DomUtils.getChildElementByTagName(stepElement, TASKLET_ELE);
if (taskletElement!=null) {
boolean stepUnderspecified = CoreNamespaceUtils.isUnderspecified(stepElement);
parseTasklet(stepElement, taskletElements.get(0), bd, parserContext, stepUnderspecified);
parseTasklet(stepElement, taskletElement, bd, parserContext, stepUnderspecified);
}
else if (taskletElements.size() > 1) {
parserContext.getReaderContext().error(
"The '<" + TASKLET_ELE + "/>' element may not appear more than once in a single <"
+ stepElement.getNodeName() + "/>.", stepElement);
Element flowElement = DomUtils.getChildElementByTagName(stepElement, FLOW_ELE);
if (flowElement!=null) {
boolean stepUnderspecified = CoreNamespaceUtils.isUnderspecified(stepElement);
parseFlow(stepElement, flowElement, bd, parserContext, stepUnderspecified);
}
String parentRef = stepElement.getAttribute(PARENT_ATTR);
@@ -103,6 +107,11 @@ public abstract class AbstractStepParser {
bd.setAbstract(Boolean.valueOf(isAbstract));
}
String jobRepositoryRef = stepElement.getAttribute(JOB_REPO_ATTR);
if (StringUtils.hasText(jobRepositoryRef)) {
builder.addPropertyReference("jobRepository", jobRepositoryRef);
}
if (StringUtils.hasText(jobFactoryRef)) {
bd.setAttribute("jobParserJobFactoryBeanRef", jobFactoryRef);
}
@@ -157,6 +166,25 @@ public abstract class AbstractStepParser {
handleTaskletElement(taskletElement, bd, parserContext);
}
private void parseFlow(Element stepElement, Element flowElement, AbstractBeanDefinition bd,
ParserContext parserContext, boolean stepUnderspecified) {
bd.setBeanClass(StepParserStepFactoryBean.class);
bd.setAttribute("isNamespaceStep", true);
String flowRef = flowElement.getAttribute(PARENT_ATTR);
String idAttribute = stepElement.getAttribute(ID_ATTR);
BeanDefinition flowDefinition = new GenericBeanDefinition();
flowDefinition.setParentName(flowRef);
MutablePropertyValues propertyValues = flowDefinition.getPropertyValues();
if (StringUtils.hasText(idAttribute)) {
propertyValues.addPropertyValue("name", idAttribute);
}
bd.getPropertyValues().addPropertyValue("flow", flowDefinition);
}
private void validateTaskletAttributesAndSubelements(Element taskletElement, ParserContext parserContext,
boolean stepUnderspecified, String taskletRef, List<Element> chunkElements, List<Element> beanElements,
List<Element> refElements) {
@@ -288,10 +316,6 @@ public abstract class AbstractStepParser {
}
private void handleTaskletAttributes(Element taskletElement, MutablePropertyValues propertyValues) {
String jobRepositoryRef = taskletElement.getAttribute(JOB_REPO_ATTR);
if (StringUtils.hasText(jobRepositoryRef)) {
propertyValues.addPropertyValue("jobRepository", new RuntimeBeanReference(jobRepositoryRef));
}
String transactionManagerRef = taskletElement.getAttribute("transaction-manager");
if (StringUtils.hasText(transactionManagerRef)) {
propertyValues.addPropertyValue("transactionManager", new RuntimeBeanReference(transactionManagerRef));

14
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/FlowElementParser.java
View File

@@ -35,22 +35,26 @@ import org.w3c.dom.Element;
*/
public class FlowElementParser {
private static final String ID_ATTR = "id";
private static final String REF_ATTR = "parent";
/**
* Parse the flow and turn it into a list of transitions.
*
* @param element the &lt;flow/gt; element to parse
* @param parserContext the parser context for the bean factory
* @return a collection of bean definitions for
* @return a collection of bean definitions for
* {@link org.springframework.batch.core.job.flow.support.StateTransition}
* instances objects
*/
public Collection<BeanDefinition> parse(Element element, ParserContext parserContext) {
String refAttribute = element.getAttribute("ref");
String idAttribute = element.getAttribute("id");
String refAttribute = element.getAttribute(REF_ATTR);
String idAttribute = element.getAttribute(ID_ATTR);
BeanDefinitionBuilder stateBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.batch.core.job.flow.support.state.FlowState");
BeanDefinitionBuilder stateBuilder = BeanDefinitionBuilder
.genericBeanDefinition("org.springframework.batch.core.job.flow.support.state.FlowState");
AbstractBeanDefinition flowDefinition = new GenericBeanDefinition();
flowDefinition.setParentName(refAttribute);

4
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SimpleFlowFactoryBean.java
View File

@@ -103,7 +103,7 @@ public class SimpleFlowFactoryBean implements FactoryBean, InitializingBean {
String stateName = prefix + oldName;
if (state instanceof StepState) {
Step step = ((StepState) state).getStep();
return new StepState(stateName, new DelegateStep(step, stateName));
return new StepState(stateName, new DelegateStep(stateName, step));
}
return new DelegateState(stateName, state);
}
@@ -154,7 +154,7 @@ public class SimpleFlowFactoryBean implements FactoryBean, InitializingBean {
private final String name;
private DelegateStep(Step step, String name) {
private DelegateStep(String name, Step step) {
this.step = step;
this.name = name;
}

6
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/SplitParser.java
View File

@@ -44,6 +44,10 @@ import org.w3c.dom.Element;
*/
public class SplitParser {
/**
*
*/
private static final String PARENT_ATTR = "parent";
private final String jobFactoryRef;
/**
@@ -91,7 +95,7 @@ public class SplitParser {
int i = 0;
String prefix = idAttribute.startsWith(jobFactoryRef) ? idAttribute : jobFactoryRef+"."+idAttribute;
for (Element nextElement : flowElements) {
String ref = nextElement.getAttribute("ref");
String ref = nextElement.getAttribute(PARENT_ATTR);
if (StringUtils.hasText(ref)) {
if (nextElement.getElementsByTagName("*").getLength() > 0) {
parserContext.getReaderContext().error("A <flow/> in a <split/> must have ref= or nested <flow/>, but not both.", nextElement);

50
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java
View File

@@ -24,6 +24,8 @@ import org.springframework.batch.classify.BinaryExceptionClassifier;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.core.job.flow.FlowStep;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean;
import org.springframework.batch.core.step.item.SimpleStepFactoryBean;
@@ -79,6 +81,11 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
private Tasklet tasklet;
private PlatformTransactionManager transactionManager;
//
// Floe Elements
//
private Flow flow;
//
// Tasklet Elements
@@ -164,6 +171,11 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
configureTaskletStep(ts);
return ts;
}
else if (flow != null) {
FlowStep ts = new FlowStep();
configureFlowStep(ts);
return ts;
}
else {
throw new IllegalStateException("Step [" + name
+ "] has neither a <chunk/> element nor a 'ref' attribute referencing a Tasklet.");
@@ -309,6 +321,33 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
}
}
@SuppressWarnings("serial")
private void configureFlowStep(FlowStep ts) {
if (name != null) {
ts.setName(name);
}
if (allowStartIfComplete != null) {
ts.setAllowStartIfComplete(allowStartIfComplete);
}
if (jobRepository != null) {
ts.setJobRepository(jobRepository);
}
if (startLimit != null) {
ts.setStartLimit(startLimit);
}
if (flow != null) {
ts.setFlow(flow);
}
if (listeners != null) {
int i = 0;
StepExecutionListener[] newListeners = new StepExecutionListener[listeners.length];
for (StepListener listener : listeners) {
newListeners[i++] = (StepExecutionListener) listener;
}
ts.setStepExecutionListeners(newListeners);
}
}
private void validateFaultTolerantSettings() {
validateDependency("skippable-exception-classes", skippableExceptionClasses, "skip-limit", skipLimit, true);
validateDependency("retryable-exception-classes", retryableExceptionClasses, "retry-limit", retryLimit, true);
@@ -388,6 +427,17 @@ class StepParserStepFactoryBean<I, O> implements FactoryBean, BeanNameAware {
this.name = name;
}
}
// =========================================================
// Flow Attributes
// =========================================================
/**
* @param flow the flow to set
*/
public void setFlow(Flow flow) {
this.flow = flow;
}
// =========================================================
// Tasklet Attributes

7
spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TopLevelFlowParser.java
View File

@@ -17,6 +17,7 @@ package org.springframework.batch.core.configuration.xml;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
@@ -27,6 +28,8 @@ public class TopLevelFlowParser extends AbstractFlowParser {
private static final String ID_ATTR = "id";
private static final String ABSTRACT_ATTR = "abstract";
/**
* @param element the top level element containing a flow definition
* @param parserContext the {@link ParserContext}
@@ -36,6 +39,10 @@ public class TopLevelFlowParser extends AbstractFlowParser {
String flowName = element.getAttribute(ID_ATTR);
builder.getRawBeanDefinition().setAttribute("flowName", flowName);
builder.addPropertyValue("name", flowName);
String abstractAttr = element.getAttribute(ABSTRACT_ATTR);
if (StringUtils.hasText(abstractAttr)) {
builder.setAbstract(abstractAttr.equals("true"));
}
super.doParse(element, parserContext, builder);
}

137
spring-batch-core/src/main/java/org/springframework/batch/core/job/AbstractJob.java
View File

@@ -27,7 +27,6 @@ import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionException;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.JobParametersIncrementer;
import org.springframework.batch.core.StartLimitExceededException;
@@ -39,7 +38,6 @@ import org.springframework.batch.core.listener.CompositeJobExecutionListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.batch.core.step.StepLocator;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.repeat.RepeatException;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
@@ -71,6 +69,8 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In
private JobParametersValidator jobParametersValidator = new DefaultJobParametersValidator();
private StepHandler stepHandler;
/**
* Default constructor.
*/
@@ -224,6 +224,16 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In
*/
public void setJobRepository(JobRepository jobRepository) {
this.jobRepository = jobRepository;
stepHandler = new SimpleStepHandler(jobRepository);
}
/**
* Convenience method for subclasses to access the job repository.
*
* @return the jobRepository
*/
protected JobRepository getJobRepository() {
return jobRepository;
}
/**
@@ -250,7 +260,7 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In
public final void execute(JobExecution execution) {
logger.debug("Job execution starting: " + execution);
try {
jobParametersValidator.validate(execution.getJobInstance().getJobParameters());
@@ -338,77 +348,8 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In
*/
protected final StepExecution handleStep(Step step, JobExecution execution) throws JobInterruptedException,
JobRestartException, StartLimitExceededException {
if (execution.isStopping()) {
throw new JobInterruptedException("JobExecution interrupted.");
}
return stepHandler.handleStep(step, execution);
JobInstance jobInstance = execution.getJobInstance();
StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step.getName());
if (stepExecutionPartOfExistingJobExecution(execution, lastStepExecution)) {
// If the last execution of this step was in the same job, it's probably
// intentional so we want to run it again...
logger.info(String.format("Duplicate step [%s] detected in execution of job=[%s]. " +
"If either step fails, both will be executed again on restart.", step.getName(), name));
lastStepExecution = null;
}
StepExecution currentStepExecution = lastStepExecution;
if (shouldStart(lastStepExecution, jobInstance, step)) {
currentStepExecution = execution.createStepExecution(step.getName());
boolean isRestart = (lastStepExecution != null && !lastStepExecution.getStatus().equals(
BatchStatus.COMPLETED));
if (isRestart) {
currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext());
}
else {
currentStepExecution.setExecutionContext(new ExecutionContext());
}
jobRepository.add(currentStepExecution);
logger.info("Executing step: [" + step + "]");
try {
step.execute(currentStepExecution);
}
catch (JobInterruptedException e) {
// Ensure that the job gets the message that it is stopping
// and can pass it on to other steps that are executing
// concurrently.
execution.setStatus(BatchStatus.STOPPING);
throw e;
}
jobRepository.updateExecutionContext(execution);
if (currentStepExecution.getStatus() == BatchStatus.STOPPING
|| currentStepExecution.getStatus() == BatchStatus.STOPPED) {
// Ensure that the job gets the message that it is stopping
execution.setStatus(BatchStatus.STOPPING);
throw new JobInterruptedException("Job interrupted by step execution");
}
}
else {
// currentStepExecution.setExitStatus(ExitStatus.NOOP);
}
return currentStepExecution;
}
/**
* Detect whether a step execution belongs to this job execution.
* @param jobExecution the current job execution
* @param stepExecution an existing step execution
* @return
*/
private boolean stepExecutionPartOfExistingJobExecution(JobExecution jobExecution, StepExecution stepExecution) {
return stepExecution != null && stepExecution.getJobExecutionId() != null
&& stepExecution.getJobExecutionId().equals(jobExecution.getId());
}
/**
@@ -419,55 +360,7 @@ public abstract class AbstractJob implements Job, StepLocator, BeanNameAware, In
* @param stepExecution
*/
protected void updateStepExecution(StepExecution stepExecution) {
jobRepository.update(stepExecution);
}
/**
* Given a step and configuration, return true if the step should start,
* false if it should not, and throw an exception if the job should finish.
* @param lastStepExecution the last step execution
* @param jobInstance
* @param step
*
* @throws StartLimitExceededException if the start limit has been exceeded
* for this step
* @throws JobRestartException if the job is in an inconsistent state from
* an earlier failure
*/
private boolean shouldStart(StepExecution lastStepExecution, JobInstance jobInstance, Step step)
throws JobRestartException, StartLimitExceededException {
BatchStatus stepStatus;
if (lastStepExecution == null) {
stepStatus = BatchStatus.STARTING;
}
else {
stepStatus = lastStepExecution.getStatus();
}
if (stepStatus == BatchStatus.UNKNOWN) {
throw new JobRestartException("Cannot restart step from UNKNOWN status. "
+ "The last execution ended with a failure that could not be rolled back, "
+ "so it may be dangerous to proceed. Manual intervention is probably necessary.");
}
if ((stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false)
|| stepStatus == BatchStatus.ABANDONED) {
// step is complete, false should be returned, indicating that the
// step should not be started
logger.info("Step already complete or not restartable, so no action to execute: " + lastStepExecution);
return false;
}
if (jobRepository.getStepExecutionCount(jobInstance, step.getName()) < step.getStartLimit()) {
// step start count is less than start max, return true
return true;
}
else {
// start max has been exceeded, throw an exception.
throw new StartLimitExceededException("Maximum start limit exceeded for step: " + step.getName()
+ "StartMax: " + step.getStartLimit());
}
stepHandler.updateStepExecution(stepExecution);
}
/**

201
spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleStepHandler.java Normal file
View File

@@ -0,0 +1,201 @@
/*
* Copyright 2006-2009 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.job;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.StartLimitExceededException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
/**
* @author Dave Syer
*
*/
public class SimpleStepHandler implements StepHandler, InitializingBean {
private static final Log logger = LogFactory.getLog(SimpleStepHandler.class);
private JobRepository jobRepository;
/**
* Convenient default constructor for configuration usage.
*/
public SimpleStepHandler() {
this(null);
}
/**
* @param jobRepository
*/
public SimpleStepHandler(JobRepository jobRepository) {
super();
this.jobRepository = jobRepository;
}
/**
* Check mandatory properties (jobRepository).
*
* @see InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.state(jobRepository!=null, "A JobRepository must be provided");
}
/**
* @param jobRepository the jobRepository to set
*/
public void setJobRepository(JobRepository jobRepository) {
this.jobRepository = jobRepository;
}
public StepExecution handleStep(Step step, JobExecution execution) throws JobInterruptedException,
JobRestartException, StartLimitExceededException {
if (execution.isStopping()) {
throw new JobInterruptedException("JobExecution interrupted.");
}
JobInstance jobInstance = execution.getJobInstance();
StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step.getName());
if (stepExecutionPartOfExistingJobExecution(execution, lastStepExecution)) {
// If the last execution of this step was in the same job, it's probably
// intentional so we want to run it again...
logger.info(String.format("Duplicate step [%s] detected in execution of job=[%s]. " +
"If either step fails, both will be executed again on restart.", step.getName(), jobInstance.getJobName()));
lastStepExecution = null;
}
StepExecution currentStepExecution = lastStepExecution;
if (shouldStart(lastStepExecution, jobInstance, step)) {
currentStepExecution = execution.createStepExecution(step.getName());
boolean isRestart = (lastStepExecution != null && !lastStepExecution.getStatus().equals(
BatchStatus.COMPLETED));
if (isRestart) {
currentStepExecution.setExecutionContext(lastStepExecution.getExecutionContext());
}
else {
currentStepExecution.setExecutionContext(new ExecutionContext());
}
jobRepository.add(currentStepExecution);
logger.info("Executing step: [" + step + "]");
try {
step.execute(currentStepExecution);
}
catch (JobInterruptedException e) {
// Ensure that the job gets the message that it is stopping
// and can pass it on to other steps that are executing
// concurrently.
execution.setStatus(BatchStatus.STOPPING);
throw e;
}
jobRepository.updateExecutionContext(execution);
if (currentStepExecution.getStatus() == BatchStatus.STOPPING
|| currentStepExecution.getStatus() == BatchStatus.STOPPED) {
// Ensure that the job gets the message that it is stopping
execution.setStatus(BatchStatus.STOPPING);
throw new JobInterruptedException("Job interrupted by step execution");
}
}
else {
// currentStepExecution.setExitStatus(ExitStatus.NOOP);
}
return currentStepExecution;
}
public void updateStepExecution(StepExecution stepExecution) {
jobRepository.update(stepExecution);
}
/**
* Detect whether a step execution belongs to this job execution.
* @param jobExecution the current job execution
* @param stepExecution an existing step execution
* @return
*/
private boolean stepExecutionPartOfExistingJobExecution(JobExecution jobExecution, StepExecution stepExecution) {
return stepExecution != null && stepExecution.getJobExecutionId() != null
&& stepExecution.getJobExecutionId().equals(jobExecution.getId());
}
/**
* Given a step and configuration, return true if the step should start,
* false if it should not, and throw an exception if the job should finish.
* @param lastStepExecution the last step execution
* @param jobInstance
* @param step
*
* @throws StartLimitExceededException if the start limit has been exceeded
* for this step
* @throws JobRestartException if the job is in an inconsistent state from
* an earlier failure
*/
private boolean shouldStart(StepExecution lastStepExecution, JobInstance jobInstance, Step step)
throws JobRestartException, StartLimitExceededException {
BatchStatus stepStatus;
if (lastStepExecution == null) {
stepStatus = BatchStatus.STARTING;
}
else {
stepStatus = lastStepExecution.getStatus();
}
if (stepStatus == BatchStatus.UNKNOWN) {
throw new JobRestartException("Cannot restart step from UNKNOWN status. "
+ "The last execution ended with a failure that could not be rolled back, "
+ "so it may be dangerous to proceed. Manual intervention is probably necessary.");
}
if ((stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false)
|| stepStatus == BatchStatus.ABANDONED) {
// step is complete, false should be returned, indicating that the
// step should not be started
logger.info("Step already complete or not restartable, so no action to execute: " + lastStepExecution);
return false;
}
if (jobRepository.getStepExecutionCount(jobInstance, step.getName()) < step.getStartLimit()) {
// step start count is less than start max, return true
return true;
}
else {
// start max has been exceeded, throw an exception.
throw new StartLimitExceededException("Maximum start limit exceeded for step: " + step.getName()
+ "StartMax: " + step.getStartLimit());
}
}
}

33
spring-batch-core/src/main/java/org/springframework/batch/core/job/StepHandler.java Normal file
View File

@@ -0,0 +1,33 @@
/*
* Copyright 2006-2009 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.job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.StartLimitExceededException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRestartException;
public interface StepHandler {
StepExecution handleStep(Step step, JobExecution execution) throws JobInterruptedException, JobRestartException,
StartLimitExceededException;
void updateStepExecution(StepExecution stepExecution);
}

95
spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java
View File

@@ -18,17 +18,12 @@ package org.springframework.batch.core.job.flow;
import java.util.Collection;
import java.util.HashSet;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionException;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.StartLimitExceededException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.AbstractJob;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.batch.core.job.SimpleStepHandler;
import org.springframework.batch.core.step.StepHolder;
/**
@@ -99,7 +94,8 @@ public class FlowJob extends AbstractJob {
@Override
protected void doExecute(final JobExecution execution) throws JobExecutionException {
try {
JobFlowExecutor executor = new JobFlowExecutor(execution);
JobFlowExecutor executor = new JobFlowExecutor(new SimpleStepHandler(getJobRepository()),
execution);
executor.updateJobExecutionStatus(flow.start(executor).getStatus());
}
catch (FlowExecutionException e) {
@@ -110,89 +106,4 @@ public class FlowJob extends AbstractJob {
}
}
/**
* @author Dave Syer
*
*/
private class JobFlowExecutor implements FlowExecutor {
private final ThreadLocal<StepExecution> stepExecutionHolder = new ThreadLocal<StepExecution>();
private final JobExecution execution;
private ExitStatus exitStatus = ExitStatus.EXECUTING;
/**
* @param execution
*/
private JobFlowExecutor(JobExecution execution) {
this.execution = execution;
stepExecutionHolder.set(null);
}
public String executeStep(Step step) throws JobInterruptedException, JobRestartException,
StartLimitExceededException {
StepExecution stepExecution = handleStep(step, execution);
stepExecutionHolder.set(stepExecution);
return stepExecution == null ? ExitStatus.COMPLETED.getExitCode() : stepExecution.getExitStatus()
.getExitCode();
}
public void abandonStepExecution() {
StepExecution lastStepExecution = stepExecutionHolder.get();
if (lastStepExecution != null && lastStepExecution.getStatus().isGreaterThan(BatchStatus.STOPPING)) {
lastStepExecution.upgradeStatus(BatchStatus.ABANDONED);
updateStepExecution(lastStepExecution);
}
}
public void updateJobExecutionStatus(FlowExecutionStatus status) {
execution.setStatus(findBatchStatus(status));
exitStatus = exitStatus.and(new ExitStatus(status.getName()));
execution.setExitStatus(exitStatus);
}
public JobExecution getJobExecution() {
return execution;
}
public StepExecution getStepExecution() {
return stepExecutionHolder.get();
}
public void close(FlowExecution result) {
stepExecutionHolder.set(null);
}
public boolean isRestart() {
if (getStepExecution() != null && getStepExecution().getStatus() == BatchStatus.ABANDONED) {
/*
* This is assumed to be the last step execution and it was
* marked abandoned, so we are in a restart of a stopped step.
* TODO: mark the step execution in some more definitive way?
*/
return true;
}
return execution.getStepExecutions().isEmpty();
}
public void addExitStatus(String code) {
exitStatus = exitStatus.and(new ExitStatus(code));
}
/**
* @param status
* @return
*/
private BatchStatus findBatchStatus(FlowExecutionStatus status) {
for (BatchStatus batchStatus : BatchStatus.values()) {
if (status.getName().startsWith(batchStatus.toString())) {
return batchStatus;
}
}
return BatchStatus.UNKNOWN;
}
}
}

48
spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowStep.java Normal file
View File

@@ -0,0 +1,48 @@
package org.springframework.batch.core.job.flow;
import org.springframework.batch.core.JobExecutionException;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.SimpleStepHandler;
import org.springframework.batch.core.job.StepHandler;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.util.Assert;
public class FlowStep extends AbstractStep {
private Flow flow;
/**
* Public setter for the flow.
*
* @param flow the flow to set
*/
public void setFlow(Flow flow) {
this.flow = flow;
}
/**
* Ensure that the flow is set.
* @see AbstractStep#afterPropertiesSet()
*/
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
Assert.state(flow!=null, "A Flow must be provided");
}
@Override
protected void doExecute(StepExecution stepExecution) throws Exception {
try {
StepHandler stepHandler = new SimpleStepHandler(getJobRepository());
FlowExecutor executor = new JobFlowExecutor(stepHandler, stepExecution.getJobExecution());
executor.updateJobExecutionStatus(flow.start(executor).getStatus());
}
catch (FlowExecutionException e) {
if (e.getCause() instanceof JobExecutionException) {
throw (JobExecutionException) e.getCause();
}
throw new JobExecutionException("Flow execution ended unexpectedly", e);
}
}
}

117
spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/JobFlowExecutor.java Normal file
View File

@@ -0,0 +1,117 @@
/*
* Copyright 2006-2009 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.job.flow;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.StartLimitExceededException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.StepHandler;
import org.springframework.batch.core.repository.JobRestartException;
/**
* Implementation of {@link FlowExecutor} for use in components that need to
* execute a flow related to a {@link JobExecution}.
*
* @author Dave Syer
*
*/
public class JobFlowExecutor implements FlowExecutor {
private final ThreadLocal<StepExecution> stepExecutionHolder = new ThreadLocal<StepExecution>();
private final JobExecution execution;
private ExitStatus exitStatus = ExitStatus.EXECUTING;
private final StepHandler stepHandler;
/**
* @param execution
*/
public JobFlowExecutor(StepHandler stepHandler, JobExecution execution) {
this.stepHandler = stepHandler;
this.execution = execution;
stepExecutionHolder.set(null);
}
public String executeStep(Step step) throws JobInterruptedException, JobRestartException,
StartLimitExceededException {
StepExecution stepExecution = stepHandler.handleStep(step, execution);
stepExecutionHolder.set(stepExecution);
return stepExecution == null ? ExitStatus.COMPLETED.getExitCode() : stepExecution.getExitStatus().getExitCode();
}
public void abandonStepExecution() {
StepExecution lastStepExecution = stepExecutionHolder.get();
if (lastStepExecution != null && lastStepExecution.getStatus().isGreaterThan(BatchStatus.STOPPING)) {
lastStepExecution.upgradeStatus(BatchStatus.ABANDONED);
stepHandler.updateStepExecution(lastStepExecution);
}
}
public void updateJobExecutionStatus(FlowExecutionStatus status) {
execution.setStatus(findBatchStatus(status));
exitStatus = exitStatus.and(new ExitStatus(status.getName()));
execution.setExitStatus(exitStatus);
}
public JobExecution getJobExecution() {
return execution;
}
public StepExecution getStepExecution() {
return stepExecutionHolder.get();
}
public void close(FlowExecution result) {
stepExecutionHolder.set(null);
}
public boolean isRestart() {
if (getStepExecution() != null && getStepExecution().getStatus() == BatchStatus.ABANDONED) {
/*
* This is assumed to be the last step execution and it was marked
* abandoned, so we are in a restart of a stopped step. TODO: mark
* the step execution in some more definitive way?
*/
return true;
}
return execution.getStepExecutions().isEmpty();
}
public void addExitStatus(String code) {
exitStatus = exitStatus.and(new ExitStatus(code));
}
/**
* @param status
* @return
*/
private BatchStatus findBatchStatus(FlowExecutionStatus status) {
for (BatchStatus batchStatus : BatchStatus.values()) {
if (status.getName().startsWith(batchStatus.toString())) {
return batchStatus;
}
}
return BatchStatus.UNKNOWN;
}
}

10
spring-batch-core/src/main/java/org/springframework/batch/core/repository/support/AbstractJobRepositoryFactoryBean.java
View File

@@ -123,6 +123,16 @@ public abstract class AbstractJobRepositoryFactoryBean implements FactoryBean, I
return transactionManager;
}
/**
* Convenience method for clients to grab the {@link JobRepository} without
* a cast.
* @return the {@link JobRepository} from {@link #getObject()}
* @throws Exception if the repository could not be created
*/
public JobRepository getJobRepository() throws Exception {
return (JobRepository) getObject();
}
private void initializeProxy() throws Exception {
if (proxyFactory == null) {
proxyFactory = new ProxyFactory();

110
spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.1.xsd
View File

@@ -110,13 +110,23 @@
<xsd:sequence>
<xsd:element ref="description" minOccurs="0" />
<xsd:choice minOccurs="0" maxOccurs="1">
<xsd:element name="tasklet">
<xsd:element name="tasklet" type="taskletType" />
<xsd:element name="flow">
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="taskletType">
<xsd:attributeGroup ref="jobRepositoryAttribute" />
</xsd:extension>
</xsd:complexContent>
<xsd:attribute name="parent" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation
source="java:org.springframework.batch.core.job.flow.Flow"><![CDATA[
The flow that will execute in this step.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="parent">
<tool:expected-type
type="org.springframework.batch.core.job.flow.Flow" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:choice>
@@ -124,6 +134,7 @@
<xsd:attribute name="id" type="xsd:ID" use="required" />
<xsd:attributeGroup ref="parentAttribute" />
<xsd:attributeGroup ref="abstractAttribute" />
<xsd:attributeGroup ref="jobRepositoryAttribute" />
</xsd:complexType>
</xsd:element>
@@ -136,12 +147,13 @@
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="description" minOccurs="0"/>
<xsd:element ref="description" minOccurs="0" />
<xsd:choice minOccurs="0" maxOccurs="unbounded">
<xsd:group ref="flowGroup" />
</xsd:choice>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID" use="required" />
<xsd:attribute name="abstract" type="xsd:boolean" />
</xsd:complexType>
</xsd:element>
@@ -285,10 +297,28 @@
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="description" minOccurs="0" />
<xsd:choice minOccurs="0" maxOccurs="unbounded">
<xsd:choice minOccurs="0" maxOccurs="1">
<xsd:element name="tasklet" type="taskletType" />
<xsd:group ref="transitions" />
<xsd:element name="flow">
<xsd:complexType>
<xsd:attribute name="parent" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation
source="java:org.springframework.batch.core.job.flow.Flow"><![CDATA[
The flow that will execute in this step.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="parent">
<tool:expected-type
type="org.springframework.batch.core.job.flow.Flow" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:choice>
<xsd:group ref="transitions" minOccurs="0" maxOccurs="unbounded" />
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID" use="required" />
<xsd:attributeGroup ref="parentAttribute" />
@@ -313,14 +343,16 @@
</xsd:annotation>
<xsd:complexType>
<xsd:group ref="flowGroup" minOccurs="0" maxOccurs="unbounded" />
<xsd:attribute name="ref" type="xsd:string" use="optional">
<xsd:attribute name="parent" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation source="java:org.springframework.batch.core.job.flow.Flow"><![CDATA[
<xsd:documentation
source="java:org.springframework.batch.core.job.flow.Flow"><![CDATA[
The flow that will execute at this point in the job.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.batch.core.job.flow.Flow" />
<tool:expected-type
type="org.springframework.batch.core.job.flow.Flow" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -351,20 +383,24 @@
<xsd:element name="flow">
<xsd:annotation>
<xsd:documentation>
Declares job should include an externalized flow here.
Declares job should include an externalized flow
here.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:group ref="transitions" minOccurs="0" maxOccurs="unbounded" />
<xsd:attribute name="id" type="xsd:ID" use="required" />
<xsd:attribute name="ref" type="xsd:string" use="required">
<xsd:attribute name="parent" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation source="java:org.springframework.batch.core.job.flow.Flow"><![CDATA[
The flow that will execute at this point in the job.
<xsd:documentation
source="java:org.springframework.batch.core.job.flow.Flow"><![CDATA[
The flow that will execute at this point in the job specified as a
parent bean definition id.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.batch.core.job.flow.Flow" />
<tool:expected-type
type="org.springframework.batch.core.job.flow.Flow" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -426,13 +462,14 @@
</xsd:element>
<xsd:element name="listeners" type="stepListenersType"
minOccurs="0" maxOccurs="1" />
<xsd:element ref="beans:bean" minOccurs="0" maxOccurs="1"/>
<xsd:element ref="beans:ref" minOccurs="0" maxOccurs="1"/>
<xsd:element ref="beans:bean" minOccurs="0" maxOccurs="1" />
<xsd:element ref="beans:ref" minOccurs="0" maxOccurs="1" />
</xsd:all>
<xsd:attribute name="ref" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
The tasklet is a reference to another bean definition that implements
The tasklet is a reference to another bean
definition that implements
the Tasklet interface.
</xsd:documentation>
<xsd:appinfo>
@@ -944,15 +981,18 @@
<xsd:annotation>
<xsd:documentation>A pattern to match against the exit status
code. Use * and ? as wildcard characters.
When a step finishes the most specific match will be chosen to
select the next step.</xsd:documentation>
When a step finishes
the most specific match will be chosen to
select the next step.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="restart" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation>The name of the step to start on when the
stopped job is restarted.
Must resolve to one of the other steps in this job.
Must resolve to one of the other steps
in this job.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -963,7 +1003,8 @@
<xsd:documentation>
Declares job should end at this point, without
the possibility of restart.
BatchStatus will be COMPLETED. ExitStatus is configurable.
BatchStatus will be COMPLETED.
ExitStatus is configurable.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
@@ -971,8 +1012,10 @@
<xsd:annotation>
<xsd:documentation>A pattern to match against the exit status
code. Use * and ? as wildcard characters.
When a step finishes the most specific match will be chosen to
select the next step.</xsd:documentation>
When a step finishes
the most specific match will be chosen to
select the next step.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="exit-code" use="optional" type="xsd:string"
@@ -996,8 +1039,10 @@
<xsd:annotation>
<xsd:documentation>A pattern to match against the exit status
code. Use * and ? as wildcard characters.
When a step finishes the most specific match will be chosen to
select the next step.</xsd:documentation>
When a step finishes
the most specific match will be chosen to
select the next step.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="exit-code" use="optional" type="xsd:string"
@@ -1058,8 +1103,10 @@
try to instantiate that particular bean
in any case.
Note: This attribute will not be inherited by child bean definitions.
Hence, it needs to be specified per abstract bean definition.
Note: This attribute will not be inherited by child
bean definitions.
Hence, it needs to be specified per abstract bean
definition.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -1071,7 +1118,8 @@
<xsd:documentation>
Should this list be merged with the corresponding
list provided
by the parent? If not, it will overwrite the parent list.
by the parent? If not, it will overwrite the parent
list.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

97
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/FlowStepParserTests.java Normal file
View File

@@ -0,0 +1,97 @@
/*
* 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.configuration.xml;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import java.util.ArrayList;
import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Dave Syer
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class FlowStepParserTests {
@Autowired
@Qualifier("job1")
private Job job1;
@Autowired
@Qualifier("job2")
private Job job2;
@Autowired
private JobRepository jobRepository;
@Autowired
private MapJobRepositoryFactoryBean mapJobRepositoryFactoryBean;
@Before
public void setUp() {
mapJobRepositoryFactoryBean.clear();
}
@Test
public void testFlowStep() throws Exception {
assertNotNull(job1);
JobExecution jobExecution = jobRepository.createJobExecution(job1.getName(), new JobParameters());
job1.execute(jobExecution);
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
List<String> stepNames = getStepNames(jobExecution);
assertEquals(5, stepNames.size());
assertEquals("[job1.s1, job1.flow, job1.flow.s2, job1.flow.s3, job1.s4]", stepNames.toString());
}
@Test
public void testFlowExternalStep() throws Exception {
assertNotNull(job2);
JobExecution jobExecution = jobRepository.createJobExecution(job1.getName(), new JobParameters());
job2.execute(jobExecution);
assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
List<String> stepNames = getStepNames(jobExecution);
assertEquals(5, stepNames.size());
assertEquals("[job2.s1, job2.flow, flow.step.s2, flow.step.s3, job2.s4]", stepNames.toString());
}
private List<String> getStepNames(JobExecution jobExecution) {
List<String> list = new ArrayList<String>();
for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
list.add(stepExecution.getStepName());
}
return list;
}
}

106
spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleStepHandlerTests.java Normal file
View File

@@ -0,0 +1,106 @@
/*
* Copyright 2006-2009 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.job;
import static org.junit.Assert.*;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.dao.StepExecutionDao;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.StepSupport;
/**
* @author Dave Syer
*
*/
public class SimpleStepHandlerTests {
private JobRepository jobRepository;
private JobExecution jobExecution;
private SimpleStepHandler stepHandler;
private StepExecutionDao stepExecutionDao;
@Before
public void setUp() throws Exception {
MapJobRepositoryFactoryBean jobRepositoryFactoryBean = new MapJobRepositoryFactoryBean();
jobRepository = jobRepositoryFactoryBean.getJobRepository();
stepExecutionDao = jobRepositoryFactoryBean.getStepExecutionDao();
jobExecution = jobRepository.createJobExecution("job", new JobParameters());
stepHandler = new SimpleStepHandler(jobRepository);
stepHandler.afterPropertiesSet();
}
/**
* Test method for {@link SimpleStepHandler#afterPropertiesSet()}.
*/
@Test(expected = IllegalStateException.class)
public void testAfterPropertiesSet() throws Exception {
SimpleStepHandler stepHandler = new SimpleStepHandler();
stepHandler.afterPropertiesSet();
}
/**
* Test method for
* {@link SimpleStepHandler#handleStep(org.springframework.batch.core.Step, org.springframework.batch.core.JobExecution)}
* .
*/
@Test
public void testHandleStep() throws Exception {
StepExecution stepExecution = stepHandler.handleStep(new StubStep("step"), jobExecution);
assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus());
}
/**
* Test method for
* {@link SimpleStepHandler#updateStepExecution(org.springframework.batch.core.StepExecution)}
*
*/
@Test
public void testUpdateStepExecution() {
StepExecution stepExecution = jobExecution.createStepExecution("step");
jobRepository.add(stepExecution);
stepExecution.setStatus(BatchStatus.FAILED);
stepHandler.updateStepExecution(stepExecution);
assertEquals(stepExecution, stepExecutionDao.getStepExecution(jobExecution, stepExecution.getId()));
}
private class StubStep extends StepSupport {
private StubStep(String name) {
super(name);
}
public void execute(StepExecution stepExecution) throws JobInterruptedException {
stepExecution.setStatus(BatchStatus.COMPLETED);
stepExecution.setExitStatus(ExitStatus.COMPLETED);
jobRepository.update(stepExecution);
}
}
}

135
spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowStepTests.java Normal file
View File

@@ -0,0 +1,135 @@
/*
* Copyright 2006-2009 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.job.flow;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import java.util.ArrayList;
import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.job.flow.support.SimpleFlow;
import org.springframework.batch.core.job.flow.support.StateTransition;
import org.springframework.batch.core.job.flow.support.state.EndState;
import org.springframework.batch.core.job.flow.support.state.StepState;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
import org.springframework.batch.core.step.StepSupport;
/**
* @author Dave Syer
*
*/
public class FlowStepTests {
private JobRepository jobRepository;
private JobExecution jobExecution;
// TODO: add XML support
@Before
public void setUp() throws Exception {
jobRepository = new MapJobRepositoryFactoryBean().getJobRepository();
jobExecution = jobRepository.createJobExecution("job", new JobParameters());
}
/**
* Test method for {@link org.springframework.batch.core.job.flow.FlowStep#afterPropertiesSet()}.
*/
@Test(expected=IllegalStateException.class)
public void testAfterPropertiesSet() throws Exception{
FlowStep step = new FlowStep();
step.setJobRepository(jobRepository);
step.afterPropertiesSet();
}
/**
* Test method for {@link org.springframework.batch.core.job.flow.FlowStep#doExecute(org.springframework.batch.core.StepExecution)}.
*/
@Test
public void testDoExecute() throws Exception {
FlowStep step = new FlowStep();
step.setJobRepository(jobRepository);
SimpleFlow flow = new SimpleFlow("job");
List<StateTransition> transitions = new ArrayList<StateTransition>();
transitions.add(StateTransition.createStateTransition(new StepState(new StubStep("step1")), "step2"));
StepState step2 = new StepState(new StubStep("step2"));
transitions.add(StateTransition.createStateTransition(step2, ExitStatus.FAILED.getExitCode(), "end0"));
transitions.add(StateTransition.createStateTransition(step2, ExitStatus.COMPLETED.getExitCode(), "end1"));
transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.FAILED, "end0")));
transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end1")));
flow.setStateTransitions(transitions);
step.setFlow(flow);
step.afterPropertiesSet();
StepExecution stepExecution = jobExecution.createStepExecution("step");
jobRepository.add(stepExecution);
step.execute(stepExecution);
stepExecution = getStepExecution(jobExecution, "step");
assertEquals(ExitStatus.COMPLETED, stepExecution.getExitStatus());
stepExecution = getStepExecution(jobExecution, "step2");
assertEquals(ExitStatus.COMPLETED, stepExecution.getExitStatus());
assertEquals(3, jobExecution.getStepExecutions().size());
}
/**
* @author Dave Syer
*
*/
private class StubStep extends StepSupport {
private StubStep(String name) {
super(name);
}
public void execute(StepExecution stepExecution) throws JobInterruptedException {
stepExecution.setStatus(BatchStatus.COMPLETED);
stepExecution.setExitStatus(ExitStatus.COMPLETED);
jobRepository.update(stepExecution);
}
}
/**
* @param jobExecution
* @param stepName
* @return the StepExecution corresponding to the specified step
*/
private StepExecution getStepExecution(JobExecution jobExecution, String stepName) {
for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
if (stepExecution.getStepName().equals(stepName)) {
return stepExecution;
}
}
fail("No stepExecution found with name: [" + stepName + "]");
return null;
}
}

10
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowJobParserTests-context.xml
View File

@@ -8,12 +8,12 @@
<job id="job1">
<step id="s1" parent="step1" next="job1.flow"/>
<flow id="job1.flow" ref="flow" next="s4" />
<flow id="job1.flow" parent="flow" next="s4" />
<step id="s4" parent="step4" />
</job>
<job id="job2">
<flow id="job2.flow" ref="flow">
<flow id="job2.flow" parent="flow">
<next on="*" to="job2.s1"/>
<next on="FAILED" to="job2.s2"/>
</flow>
@@ -22,13 +22,13 @@
</job>
<job id="job3">
<flow id="job3.flow" ref="flow" />
<flow id="job3.flow" parent="flow" />
</job>
<job id="job4">
<split id="split">
<flow ref="flow" />
<flow ref="flow" />
<flow parent="flow" />
<flow parent="flow" />
</split>
</job>

32
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/FlowStepParserTests-context.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/batch"
xmlns:beans="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch-2.1.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd">
<beans:import resource="common-context.xml" />
<job id="job1">
<step id="s1" parent="step1" next="job1.flow" />
<step id="job1.flow" next="s4">
<flow parent="flow" />
</step>
<step id="s4" parent="step4" />
</job>
<job id="job2">
<step id="job2.s1" parent="step1" next="job2.flow" />
<step id="job2.flow" parent="flow.step" next="job2.s4" />
<step id="job2.s4" parent="step4" />
</job>
<flow id="flow" abstract="true">
<step id="s2" parent="step2" next="s3" />
<step id="s3" parent="step3" />
</flow>
<step id="flow.step" abstract="true">
<flow parent="flow" />
</step>
</beans:beans>

4
spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepParserParentAttributeTests-context.xml
View File

@@ -127,8 +127,8 @@
<tasklet ref="dummyTasklet"/>
</step>
<step id="specifiedRepoStandaloneStep">
<tasklet job-repository="dummyJobRepository2" transaction-manager="dummyTxMgr" ref="dummyTasklet"/>
<step id="specifiedRepoStandaloneStep" job-repository="dummyJobRepository2">
<tasklet transaction-manager="dummyTxMgr" ref="dummyTasklet"/>
</step>
<job id="baseJobWithRepo" job-repository="dummyJobRepository2" abstract="true"/>

4
spring-batch-samples/src/main/resources/jobs/partitionFileJob.xml
View File

@@ -34,8 +34,8 @@
</property>
</bean>
<step id="step1" xmlns="http://www.springframework.org/schema/batch">
<tasklet job-repository="jobRepository" transaction-manager="transactionManager">
<step id="step1" xmlns="http://www.springframework.org/schema/batch" job-repository="jobRepository">
<tasklet transaction-manager="transactionManager">
<chunk writer="itemWriter" reader="itemReader" processor="itemProcessor" commit-interval="5" />
<listeners>
<listener ref="fileNameListener" />