diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractFlowParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractFlowParser.java index 83e393624..9f91af2f1 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractFlowParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractFlowParser.java @@ -15,6 +15,15 @@ */ package org.springframework.batch.core.configuration.xml; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + import org.springframework.batch.core.job.flow.FlowExecutionStatus; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.parsing.CompositeComponentDefinition; @@ -28,8 +37,6 @@ import org.w3c.dom.Element; import org.w3c.dom.Node; import org.w3c.dom.NodeList; -import java.util.*; - /** * @author Dave Syer * @@ -182,14 +189,12 @@ public abstract class AbstractFlowParser extends AbstractSingleBeanDefinitionPar reachableElements.add(nextAttribute); } - @SuppressWarnings("unchecked") List nextElements = DomUtils.getChildElementsByTagName(element, NEXT_ELE); for (Element nextElement : nextElements) { String toAttribute = nextElement.getAttribute(TO_ATTR); reachableElements.add(toAttribute); } - @SuppressWarnings("unchecked") List stopElements = DomUtils.getChildElementsByTagName(element, STOP_ELE); for (Element stopElement : stopElements) { String restartAttribute = stopElement.getAttribute(RESTART_ATTR); @@ -257,7 +262,6 @@ public abstract class AbstractFlowParser extends AbstractSingleBeanDefinitionPar boolean transitionElementExists = false; List patterns = new ArrayList(); for (String transitionName : new String[] { NEXT_ELE, STOP_ELE, END_ELE, FAIL_ELE }) { - @SuppressWarnings("unchecked") List transitionElements = DomUtils.getChildElementsByTagName(element, transitionName); for (Element transitionElement : transitionElements) { verifyUniquePattern(transitionElement, patterns, element, parserContext); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowBuilder.java new file mode 100644 index 000000000..dd06f9a97 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowBuilder.java @@ -0,0 +1,557 @@ +/* + * Copyright 2012-2013 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.builder; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.job.flow.Flow; +import org.springframework.batch.core.job.flow.FlowExecutionStatus; +import org.springframework.batch.core.job.flow.JobExecutionDecider; +import org.springframework.batch.core.job.flow.State; +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.DecisionState; +import org.springframework.batch.core.job.flow.support.state.EndState; +import org.springframework.batch.core.job.flow.support.state.FlowState; +import org.springframework.batch.core.job.flow.support.state.SplitState; +import org.springframework.batch.core.job.flow.support.state.StepState; +import org.springframework.core.task.TaskExecutor; + +/** + * A builder for a flow of steps that can be executed as a job or as part of a job. Steps can be linked together with + * conditional transitions that depend on the exit status of the previous step. + * + * @author Dave Syer + * + * @since 2.2 + * + * @param the type of object returned by the builder (by default a Flow) + * + */ +public class FlowBuilder { + + private String name; + + private String prefix; + + private List transitions = new ArrayList(); + + private Map tos = new HashMap(); + + private State currentState; + + private EndState failedState; + + private EndState completedState; + + private int decisionCounter = 0; + + private int splitCounter = 0; + + private Map states = new HashMap(); + + private SimpleFlow flow; + + private boolean dirty = true; + + public FlowBuilder(String name) { + this.name = name; + this.prefix = name + "."; + this.failedState = new EndState(FlowExecutionStatus.FAILED, prefix + "FAILED"); + this.completedState = new EndState(FlowExecutionStatus.COMPLETED, prefix + "COMPLETED"); + } + + /** + * Validate the current state of the builder and build a flow. Subclasses may override this to build an object of a + * different type that itself depends on the flow. + * + * @return a flow + */ + public Q build() { + @SuppressWarnings("unchecked") + Q result = (Q) flow(); + return result; + } + + /** + * Transition to the next step on successful completion of the current step. All other outcomes are treated as + * failures. + * + * @param step the next step + * @return this to enable chaining + */ + public FlowBuilder next(Step step) { + doNext(step); + return this; + } + + /** + * Start a flow. If some steps are already registered, just a synonym for {@link #from(Step)}. + * + * @param step the step to start with + * @return this to enable chaining + */ + public FlowBuilder start(Step step) { + doStart(step); + return this; + } + + /** + * Go back to a previously registered step and start a new path. If no steps are registered yet just a synonym for + * {@link #start(Step)}. + * + * @param step the step to start from (already registered) + * @return this to enable chaining + */ + public FlowBuilder from(Step step) { + doFrom(step); + return this; + } + + /** + * Transition to the decider on successful completion of the current step. All other outcomes are treated as + * failures. + * + * @param step the next step + * @return this to enable chaining + */ + public UnterminatedFlowBuilder next(JobExecutionDecider decider) { + doNext(decider); + return new UnterminatedFlowBuilder(this); + } + + /** + * If a flow should start with a decision use this as the first state. + * + * @param decider the to start from + * @return a builder to enable chaining + */ + public UnterminatedFlowBuilder start(JobExecutionDecider decider) { + doStart(decider); + return new UnterminatedFlowBuilder(this); + } + + /** + * Start again from a decision that was already registered. + * + * @param decider the decider to start from (already registered) + * @return a builder to enable chaining + */ + public UnterminatedFlowBuilder from(JobExecutionDecider decider) { + doFrom(decider); + return new UnterminatedFlowBuilder(this); + } + + /** + * Go next on successful completion to a subflow. + * + * @param flow the flow to go to + * @return a builder to enable chaining + */ + public FlowBuilder next(Flow flow) { + doNext(flow); + return this; + } + + /** + * Start again from a subflow that was already registered. + * + * @param flow the flow to start from (already registered) + * @return a builder to enable chaining + */ + public FlowBuilder from(Flow flow) { + doFrom(flow); + return this; + } + + /** + * If a flow should start with a subflow use this as the first state. + * + * @param flow the flow to start from + * @return a builder to enable chaining + */ + public FlowBuilder start(Flow flow) { + doStart(flow); + return this; + } + + /** + * @param executor a task executor to execute the split flows + * @return a builder to enable fluent chaining + */ + public SplitBuilder split(TaskExecutor executor) { + return new SplitBuilder(this, executor); + } + + /** + * Start a transition to a new state if the exit status from the previous state matches the pattern given. + * Successful completion normally results in an exit status equal to (or starting with by convention) "COMPLETED". + * See {@link ExitStatus} for commonly used values. + * + * @param pattern the pattern of exit status on which to take this transition + * @return a builder to enable fluent chaining + */ + public TransitionBuilder on(String pattern) { + return new TransitionBuilder(this, pattern); + } + + /** + * A synonym for {@link #build()} which callers might find useful. Subclasses can override build to create an object + * of the desired type (e.g. a parent builder or an actual flow). + * + * @return the result of the builder + * + * @param Q the type of the result + */ + public final Q end() { + return build(); + } + + protected Flow flow() { + if (!dirty) { + // optimization in case this method is called consecutively + return flow; + } + flow = new SimpleFlow(name); + // optimization for flows that only have one state that itself is a flow: + if (currentState instanceof FlowState && states.size() == 1) { + return ((FlowState) currentState).getFlows().iterator().next(); + } + addDanglingEndStates(); + flow.setStateTransitions(transitions); + dirty = false; + return flow; + } + + private void doNext(Object input) { + if (this.currentState == null) { + doStart(input); + } + State next = createState(input); + addTransition("COMPLETED", next); + addTransition("*", failedState); + this.currentState = next; + } + + private void doStart(Object input) { + if (this.currentState != null) { + doFrom(input); + } + this.currentState = createState(input); + } + + private void doFrom(Object input) { + if (currentState == null) { + doStart(input); + } + State state = createState(input); + tos.put(currentState.getName(), currentState); + this.currentState = state; + } + + private State createState(Object input) { + State result; + if (input instanceof Step) { + if (!states.containsKey(input)) { + Step step = (Step) input; + states.put(input, new StepState(prefix + step.getName(), step)); + } + result = states.get(input); + } + else if (input instanceof JobExecutionDecider) { + if (!states.containsKey(input)) { + states.put(input, new DecisionState((JobExecutionDecider) input, prefix + "decision" + + (decisionCounter++))); + } + result = states.get(input); + } + else if (input instanceof Flow) { + if (!states.containsKey(input)) { + states.put(input, new FlowState((Flow) input, prefix + ((Flow) input).getName())); + } + result = states.get(input); + } + else { + throw new FlowBuilderException("No state can be created for: " + input); + } + dirty = true; + return result; + } + + private SplitState createState(Collection flows, TaskExecutor executor) { + if (!states.containsKey(flows)) { + states.put(flows, new SplitState(flows, prefix + "split" + (splitCounter++))); + } + SplitState result = (SplitState) states.get(flows); + if (executor != null) { + result.setTaskExecutor(executor); + } + dirty = true; + return result; + } + + private void addDanglingEndStates() { + Set froms = new HashSet(); + for (StateTransition transition : transitions) { + froms.add(transition.getState().getName()); + } + if (tos.isEmpty() && currentState != null) { + tos.put(currentState.getName(), currentState); + } + Map copy = new HashMap(tos); + // Find all the states that are really end states but not explicitly declared as such + for (String to : copy.keySet()) { + if (!froms.contains(to)) { + currentState = copy.get(to); + if (currentState != completedState) { + addTransition("COMPLETED", completedState); + } + if (currentState != failedState) { + addTransition("*", failedState); + } + } + } + copy = new HashMap(tos); + // Then find the states that do not have a default transition + for (String from : copy.keySet()) { + currentState = copy.get(from); + if (currentState != failedState) { + if (!hasFail(from)) { + addTransition("*", failedState); + } + } + if (currentState != completedState) { + if (!hasCompleted(from)) { + addTransition("*", completedState); + } + } + } + } + + private boolean hasFail(String from) { + return matches(from, "FAILED"); + } + + private boolean hasCompleted(String from) { + return matches(from, "COMPLETED"); + } + + private boolean matches(String from, String status) { + for (StateTransition transition : transitions) { + if (from.equals(transition.getState().getName()) && transition.matches(status)) { + return true; + } + } + return false; + } + + private void addTransition(String pattern, State next) { + tos.put(next.getName(), next); + transitions.add(StateTransition.createStateTransition(currentState, pattern, next.getName())); + if (transitions.size() == 1) { + transitions.add(StateTransition.createEndStateTransition(failedState)); + transitions.add(StateTransition.createEndStateTransition(completedState)); + } + dirty = true; + } + + private void end(String pattern) { + addTransition(pattern, completedState); + } + + private void fail(String pattern) { + addTransition(pattern, failedState); + } + + /** + * A builder for continuing a flow from a decision state. + * + * @author Dave Syer + * + * @param the result of the builder's build() + */ + public static class UnterminatedFlowBuilder { + + private final FlowBuilder parent; + + public UnterminatedFlowBuilder(FlowBuilder parent) { + this.parent = parent; + } + + /** + * Start a transition to a new state if the exit status from the previous state matches the pattern given. + * Successful completion normally results in an exit status equal to (or starting with by convention) + * "COMPLETED". See {@link ExitStatus} for commonly used values. + * + * @param pattern the pattern of exit status on which to take this transition + * @return + */ + public TransitionBuilder on(String pattern) { + return new TransitionBuilder(parent, pattern); + } + + } + + /** + * A builder for transitions within a flow. + * + * @author Dave Syer + * + * @param the result of the parent builder's build() + */ + public static class TransitionBuilder { + + private final FlowBuilder parent; + + private final String pattern; + + public TransitionBuilder(FlowBuilder parent, String pattern) { + this.parent = parent; + this.pattern = pattern; + } + + /** + * Specify the next step. + * + * @param step the next step after this transition + * @return a FlowBuilder + */ + public FlowBuilder to(Step step) { + State next = parent.createState(step); + parent.addTransition(pattern, next); + parent.currentState = next; + return parent; + } + + /** + * Specify the next state as a complete flow. + * + * @param step the next step after this transition + * @return a FlowBuilder + */ + public FlowBuilder to(Flow flow) { + State next = parent.createState(flow); + parent.addTransition(pattern, next); + parent.currentState = next; + return parent; + } + + /** + * Specify the next state as a decision. + * + * @param step the next step after this transition + * @return a FlowBuilder + */ + public FlowBuilder to(JobExecutionDecider decider) { + State next = parent.createState(decider); + parent.addTransition(pattern, next); + parent.currentState = next; + return parent; + } + + /** + * Signal the successful end of the flow. + * + * @return a FlowBuilder + */ + public FlowBuilder end() { + parent.end(pattern); + return parent; + } + + /** + * Signal the end of the flow with an error condition. + * + * @return a FlowBuilder + */ + public FlowBuilder fail() { + parent.fail(pattern); + return parent; + } + } + + /** + * A builder for building a split state. Example (builder is a {@link FlowBuilder}): + * + * 
+	 * Flow splitFlow = builder.start(flow1).split(new SyncTaskExecutor()).add(flow2).build();
+	 *
+ * + * where flow1 and flow2 will be executed (one after the other because of the task + * executor that was added). Another example + * + * 
+	 * Flow splitFlow = builder.start(step1).split(new SimpleAsyncTaskExecutor()).add(flow).build();
+	 *
+ * + * In this example, a flow consisting of step1 will be executed in parallel with flow. + * + * @author Dave Syer + * + * @param the result of the parent builder's build() + */ + public static class SplitBuilder { + + private final FlowBuilder parent; + + private TaskExecutor executor; + + /** + * @param parent the parent builder + * @param executor the task executor to use in the split + */ + public SplitBuilder(FlowBuilder parent, TaskExecutor executor) { + this.parent = parent; + this.executor = executor; + } + + /** + * Add flows to the split, in addition to the current state already present in the parent builder. + * + * @param flows more flows to add to the split + * @return the parent builder + */ + public FlowBuilder add(Flow... flows) { + Collection list = new ArrayList(Arrays.asList(flows)); + String name = "split" + (parent.splitCounter++); + int counter = 0; + State one = parent.currentState; + Flow flow = null; + if (!(one instanceof FlowState)) { + FlowBuilder stateBuilder = new FlowBuilder(name + "_" + (counter++)); + stateBuilder.currentState = one; + flow = stateBuilder.build(); + } + if (flow != null) { + list.add(flow); + } + State next = parent.createState(list, executor); + parent.currentState = next; + return parent; + } + + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowBuilderException.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowBuilderException.java new file mode 100644 index 000000000..7a065b5ff --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowBuilderException.java @@ -0,0 +1,38 @@ +/* + * Copyright 2012-2013 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.builder; + +/** + * @author Dave Syer + * + * @since 2.2 + * + */ +public class FlowBuilderException extends RuntimeException { + + public FlowBuilderException(String msg, Exception e) { + super(msg, e); + } + + public FlowBuilderException(Exception e) { + super(e); + } + + public FlowBuilderException(String msg) { + super(msg); + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowJobBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowJobBuilder.java new file mode 100644 index 000000000..3ada9d748 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/FlowJobBuilder.java @@ -0,0 +1,95 @@ +/* + * Copyright 2006-2011 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.builder; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.job.flow.Flow; +import org.springframework.batch.core.job.flow.FlowJob; +import org.springframework.batch.core.step.builder.StepBuilderException; + +/** + * A job builder for {@link FlowJob} instances. A flow job delegates processing to a nested flow composed of steps and + * conditional transitions between steps. + * + * @author Dave Syer + * + * @since 2.2 + */ +public class FlowJobBuilder extends JobBuilderHelper { + + private Flow flow; + + /** + * Create a new builder initialized with any properties in the parent. The parent is copied, so it can be re-used. + * + * @param parent a parent helper containing common job properties + */ + public FlowJobBuilder(JobBuilderHelper parent) { + super(parent); + } + + /** + * Start a job with this flow, but expect to transition from there to other flows or steps. + * + * @param flow the flow to start with + * @return a builder to enable fluent chaining + */ + public JobFlowBuilder start(Flow flow) { + return new JobFlowBuilder(this, flow); + } + + /** + * Start a job with this step, but expect to transition from there to other flows or steps. + * + * @param step the step to start with + * @return a builder to enable fluent chaining + */ + public JobFlowBuilder start(Step step) { + return new JobFlowBuilder(this, step); + } + + /** + * Provide a single flow to execute as the job. + * + * @param flow the flow to execute + * @return this for fluent chaining + */ + protected FlowJobBuilder flow(Flow flow) { + this.flow = flow; + return this; + } + + /** + * Build a job that executes the flow provided, normally composed of other steps. + * + * @return a flow job + */ + public Job build() { + FlowJob job = new FlowJob(); + job.setName(getName()); + job.setFlow(flow); + super.enhance(job); + try { + job.afterPropertiesSet(); + } + catch (Exception e) { + throw new StepBuilderException(e); + } + return job; + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilder.java new file mode 100644 index 000000000..37c608880 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilder.java @@ -0,0 +1,71 @@ +/* + * Copyright 2006-2011 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.builder; + +import org.springframework.batch.core.Step; +import org.springframework.batch.core.job.flow.Flow; + +/** + * Convenience for building jobs of various kinds. + * + * @author Dave Syer + * + * @since 2.2 + * + */ +public class JobBuilder extends JobBuilderHelper { + + /** + * Create a new builder for a job with the given name. + * + * @param name the name of the job + */ + public JobBuilder(String name) { + super(name); + } + + /** + * Create a new job builder that will execute a step or sequence of steps. + * + * @param step a step to execute + * @return a {@link SimpleJobBuilder} + */ + public SimpleJobBuilder start(Step step) { + return new SimpleJobBuilder(this).start(step); + } + + /** + * Create a new job builder that will execute a step or sequence of steps. + * + * @param step a step to execute + * @return a {@link SimpleJobBuilder} + */ + public JobFlowBuilder start(Flow flow) { + return new FlowJobBuilder(this).start(flow); + } + + /** + * Create a new job builder that will execute a step or sequence of steps. + * + * @param step a step to execute + * @return a {@link SimpleJobBuilder} + */ + public JobFlowBuilder flow(Step step) { + return new FlowJobBuilder(this).start(step); + } + + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderException.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderException.java new file mode 100644 index 000000000..c59ff9757 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderException.java @@ -0,0 +1,30 @@ +/* + * Copyright 2012-2013 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.builder; + +/** + * @author Dave Syer + * + * @since 2.2 + * + */ +public class JobBuilderException extends RuntimeException { + + public JobBuilderException(Exception e) { + super(e); + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderHelper.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderHelper.java new file mode 100644 index 000000000..b4b3649de --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobBuilderHelper.java @@ -0,0 +1,242 @@ +/* + * Copyright 2006-2011 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.builder; + +import java.util.ArrayList; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecutionListener; +import org.springframework.batch.core.JobParametersIncrementer; +import org.springframework.batch.core.JobParametersValidator; +import org.springframework.batch.core.job.AbstractJob; +import org.springframework.batch.core.repository.JobRepository; + +/** + * A base class and utility for other job builders providing access to common properties like job repository. + * + * @author Dave Syer + * + * @since 2.2 + */ +public abstract class JobBuilderHelper> { + + protected final Log logger = LogFactory.getLog(getClass()); + + private final CommonStepProperties properties; + + public JobBuilderHelper(String name) { + this.properties = new CommonStepProperties(); + properties.name = name; + } + + /** + * Create a new builder initialized with any properties in the parent. The parent is copied, so it can be re-used. + * + * @param parent a parent helper containing common step properties + */ + protected JobBuilderHelper(JobBuilderHelper parent) { + this.properties = new CommonStepProperties(parent.properties); + } + + /** + * Add a job parameters validator. + * + * @param jobParametersValidator a job parameters validator + * @return this to enable fluent chaining + */ + public B validator(JobParametersValidator jobParametersValidator) { + properties.jobParametersValidator = jobParametersValidator; + @SuppressWarnings("unchecked") + B result = (B) this; + return result; + } + + /** + * Add a job parameters incrementer. + * + * @param jobParametersIncrementer a job parameters incrementer + * @return this to enable fluent chaining + */ + public B incrementer(JobParametersIncrementer jobParametersIncrementer) { + properties.jobParametersIncrementer = jobParametersIncrementer; + @SuppressWarnings("unchecked") + B result = (B) this; + return result; + } + + /** + * Sets the job repository for the job. + * + * @param jobRepository the job repository (mandatory) + * @return this to enable fluent chaining + */ + public B repository(JobRepository jobRepository) { + properties.jobRepository = jobRepository; + @SuppressWarnings("unchecked") + B result = (B) this; + return result; + } + + /** + * Register a job execution listener. + * + * @param listener a job execution listener + * @return this to enable fluent chaining + */ + public B listener(JobExecutionListener listener) { + properties.addJobExecutionListener(listener); + @SuppressWarnings("unchecked") + B result = (B) this; + return result; + } + + /** + * Set a flag to prevent restart an execution of this job even if it has failed. + * + * @return this to enable fluent chaining + */ + public B preventRestart() { + properties.restartable = false; + @SuppressWarnings("unchecked") + B result = (B) this; + return result; + } + + protected String getName() { + return properties.name; + } + + protected JobRepository getJobRepository() { + return properties.jobRepository; + } + + protected boolean isRestartable() { + return properties.restartable; + } + + protected void enhance(Job target) { + + if (target instanceof AbstractJob) { + + AbstractJob job = (AbstractJob) target; + job.setJobRepository(properties.getJobRepository()); + + JobParametersIncrementer jobParametersIncrementer = properties.getJobParametersIncrementer(); + if (jobParametersIncrementer != null) { + job.setJobParametersIncrementer(jobParametersIncrementer); + } + JobParametersValidator jobParametersValidator = properties.getJobParametersValidator(); + if (jobParametersValidator != null) { + job.setJobParametersValidator(jobParametersValidator); + } + + Boolean restartable = properties.getRestartable(); + if (restartable != null) { + job.setRestartable(restartable); + } + + List listeners = properties.getJobExecutionListeners(); + if (!listeners.isEmpty()) { + job.setJobExecutionListeners(listeners.toArray(new JobExecutionListener[0])); + } + + } + + } + + public static class CommonStepProperties { + + private Set jobExecutionListeners = new LinkedHashSet(); + + private boolean restartable = true; + + private JobRepository jobRepository; + + private JobParametersIncrementer jobParametersIncrementer; + + private JobParametersValidator jobParametersValidator; + + public CommonStepProperties() { + } + + public CommonStepProperties(CommonStepProperties properties) { + this.name = properties.name; + this.restartable = properties.restartable; + this.jobRepository = properties.jobRepository; + } + + public JobParametersIncrementer getJobParametersIncrementer() { + return jobParametersIncrementer; + } + + public void setJobParametersIncrementer(JobParametersIncrementer jobParametersIncrementer) { + this.jobParametersIncrementer = jobParametersIncrementer; + } + + public JobParametersValidator getJobParametersValidator() { + return jobParametersValidator; + } + + public void setJobParametersValidator(JobParametersValidator jobParametersValidator) { + this.jobParametersValidator = jobParametersValidator; + } + + public JobRepository getJobRepository() { + return jobRepository; + } + + public void setJobRepository(JobRepository jobRepository) { + this.jobRepository = jobRepository; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public List getJobExecutionListeners() { + return new ArrayList(jobExecutionListeners); + } + + public void addStepExecutionListeners(List jobExecutionListeners) { + this.jobExecutionListeners.addAll(jobExecutionListeners); + } + + public void addJobExecutionListener(JobExecutionListener jobExecutionListener) { + this.jobExecutionListeners.add(jobExecutionListener); + } + + public boolean getRestartable() { + return restartable; + } + + public void setRestartable(boolean restartable) { + this.restartable = restartable; + } + + private String name; + + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobFlowBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobFlowBuilder.java new file mode 100644 index 000000000..55e7ec971 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/JobFlowBuilder.java @@ -0,0 +1,65 @@ +/* + * Copyright 2012-2013 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.builder; + +import org.springframework.batch.core.Step; +import org.springframework.batch.core.job.flow.Flow; +import org.springframework.batch.core.job.flow.JobExecutionDecider; + +/** + * @author Dave Syer + * + */ +public class JobFlowBuilder extends FlowBuilder { + + private FlowJobBuilder parent; + + public JobFlowBuilder(FlowJobBuilder parent) { + super(parent.getName()); + this.parent = parent; + } + + public JobFlowBuilder(FlowJobBuilder parent, Step step) { + super(parent.getName()); + this.parent = parent; + start(step); + } + + public JobFlowBuilder(FlowJobBuilder parent, JobExecutionDecider decider) { + super(parent.getName()); + this.parent = parent; + start(decider); + } + + public JobFlowBuilder(FlowJobBuilder parent, Flow flow) { + super(parent.getName()); + this.parent = parent; + start(flow); + } + + /** + * Build a flow and inject it into the parent builder. The parent builder is then returned so it can be enhanced + * before building an actual job. Normally called explicitly via {@link #end()}. + * + * @see org.springframework.batch.core.job.builder.FlowBuilder#build() + */ + public FlowJobBuilder build() { + Flow flow = flow(); + parent.flow(flow); + return parent; + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/SimpleJobBuilder.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/SimpleJobBuilder.java new file mode 100644 index 000000000..10a119922 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/builder/SimpleJobBuilder.java @@ -0,0 +1,178 @@ +/* + * Copyright 2012-2013 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.builder; + +import java.util.ArrayList; +import java.util.List; + +import org.springframework.batch.core.Job; +import org.springframework.batch.core.Step; +import org.springframework.batch.core.job.SimpleJob; +import org.springframework.batch.core.job.flow.JobExecutionDecider; +import org.springframework.core.task.TaskExecutor; +import org.springframework.util.Assert; + +/** + * @author Dave Syer + * + * @since 2.2 + * + */ +public class SimpleJobBuilder extends JobBuilderHelper { + + private List steps = new ArrayList(); + + private JobFlowBuilder builder; + + /** + * Create a new builder initialized with any properties in the parent. The parent is copied, so it can be re-used. + * + * @param parent the parent to use + */ + public SimpleJobBuilder(JobBuilderHelper parent) { + super(parent); + } + + public Job build() { + if (builder != null) { + return builder.end().build(); + } + SimpleJob job = new SimpleJob(); + super.enhance(job); + job.setSteps(steps); + try { + job.afterPropertiesSet(); + } + catch (Exception e) { + throw new JobBuilderException(e); + } + return job; + } + + /** + * Start the job with this step. + * + * @param step a step to start with + * @return this for fluent chaining + */ + public SimpleJobBuilder start(Step step) { + if (steps.isEmpty()) { + steps.add(step); + } + else { + steps.set(0, step); + } + return this; + } + + /** + * Branch into a flow conditional on the outcome of the current step. + * + * @param pattern a pattern for the exit status of the current step + * @return a builder for fluent chaining + */ + public FlowBuilder.TransitionBuilder on(String pattern) { + Assert.state(steps.size() > 0, "You have to start a job with a step"); + for (Step step : steps) { + if (builder == null) { + builder = new JobFlowBuilder(new FlowJobBuilder(this), step); + } + else { + builder.next(step); + } + } + return builder.on(pattern); + } + + /** + * Start with this decider. Returns a flow builder and when the flow is ended a job builder will be returned to + * continue the job configuration if needed. + * + * @param decider a decider to execute first + * @return builder for fluent chaining + */ + public JobFlowBuilder start(JobExecutionDecider decider) { + if (builder == null) { + builder = new JobFlowBuilder(new FlowJobBuilder(this), decider); + } + else { + builder.start(decider); + } + if (!steps.isEmpty()) { + steps.remove(0); + } + for (Step step : steps) { + builder.next(step); + } + return builder; + } + + /** + * Continue with this decider if the previous step was successful. Returns a flow builder and when the flow is ended + * a job builder will be returned to continue the job configuration if needed. + * + * @param decider a decider to execute next + * @return builder for fluent chaining + */ + public JobFlowBuilder next(JobExecutionDecider decider) { + for (Step step : steps) { + if (builder == null) { + builder = new JobFlowBuilder(new FlowJobBuilder(this), step); + } + else { + builder.next(step); + } + } + if (builder == null) { + builder = new JobFlowBuilder(new FlowJobBuilder(this), decider); + } + else { + builder.next(decider); + } + return builder; + } + + /** + * Continue or end a job with this step if the previous step was successful. + * + * @param step a step to execute next + * @return this for fluent chaining + */ + public SimpleJobBuilder next(Step step) { + steps.add(step); + return this; + } + + /** + * @param executor + * @return builder for fluent chaining + */ + public JobFlowBuilder.SplitBuilder split(TaskExecutor executor) { + for (Step step : steps) { + if (builder == null) { + builder = new JobFlowBuilder(new FlowJobBuilder(this), step); + } + else { + builder.next(step); + } + } + if (builder == null) { + builder = new JobFlowBuilder(new FlowJobBuilder(this)); + } + return builder.split(executor); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/builder/FlowBuilderTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/builder/FlowBuilderTests.java new file mode 100644 index 000000000..067c09485 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/builder/FlowBuilderTests.java @@ -0,0 +1,50 @@ +/* + * Copyright 2012-2013 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.builder; + +import org.junit.Test; +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.UnexpectedJobExecutionException; +import org.springframework.batch.core.job.SimpleStepHandler; +import org.springframework.batch.core.job.flow.Flow; +import org.springframework.batch.core.job.flow.JobFlowExecutor; +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 FlowBuilderTests { + + @Test + public void test() throws Exception { + FlowBuilder builder = new FlowBuilder("flow"); + JobRepository jobRepository = new MapJobRepositoryFactoryBean().getJobRepository(); + JobExecution execution = jobRepository.createJobExecution("foo", new JobParameters()); + builder.start(new StepSupport("step") { + @Override + public void execute(StepExecution stepExecution) throws JobInterruptedException, + UnexpectedJobExecutionException { + } + }).end().start(new JobFlowExecutor(jobRepository, new SimpleStepHandler(jobRepository), execution)); + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/builder/FlowJobBuilderTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/builder/FlowJobBuilderTests.java new file mode 100644 index 000000000..59b9a4481 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/builder/FlowJobBuilderTests.java @@ -0,0 +1,167 @@ +/* + * Copyright 2012-2013 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.builder; + +import static org.junit.Assert.assertEquals; + +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.UnexpectedJobExecutionException; +import org.springframework.batch.core.job.flow.Flow; +import org.springframework.batch.core.job.flow.FlowExecutionStatus; +import org.springframework.batch.core.job.flow.JobExecutionDecider; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; +import org.springframework.batch.core.step.StepSupport; +import org.springframework.core.task.SimpleAsyncTaskExecutor; + +/** + * @author Dave Syer + * + */ +public class FlowJobBuilderTests { + + private JobRepository jobRepository; + + private JobExecution execution; + + private StepSupport step1 = new StepSupport("step1") { + @Override + public void execute(StepExecution stepExecution) throws JobInterruptedException, + UnexpectedJobExecutionException { + stepExecution.upgradeStatus(BatchStatus.COMPLETED); + stepExecution.setExitStatus(ExitStatus.COMPLETED); + jobRepository.update(stepExecution); + } + }; + + private StepSupport step2 = new StepSupport("step2") { + @Override + public void execute(StepExecution stepExecution) throws JobInterruptedException, + UnexpectedJobExecutionException { + stepExecution.upgradeStatus(BatchStatus.COMPLETED); + stepExecution.setExitStatus(ExitStatus.COMPLETED); + jobRepository.update(stepExecution); + } + }; + + private StepSupport step3 = new StepSupport("step3") { + @Override + public void execute(StepExecution stepExecution) throws JobInterruptedException, + UnexpectedJobExecutionException { + stepExecution.upgradeStatus(BatchStatus.COMPLETED); + stepExecution.setExitStatus(ExitStatus.COMPLETED); + jobRepository.update(stepExecution); + } + }; + + @Before + public void init() throws Exception { + jobRepository = new MapJobRepositoryFactoryBean().getJobRepository(); + execution = jobRepository.createJobExecution("flow", new JobParameters()); + } + + @Test + public void testBuildOnOneLine() throws Exception { + FlowJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step1).on("COMPLETED") + .to(step2).end().preventRestart(); + builder.build().execute(execution); + assertEquals(BatchStatus.COMPLETED, execution.getStatus()); + assertEquals(2, execution.getStepExecutions().size()); + } + + @Test + public void testBuildSingleFlow() throws Exception { + Flow flow = new FlowBuilder("subflow").from(step1).next(step2).build(); + FlowJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(flow).end().preventRestart(); + builder.build().execute(execution); + assertEquals(BatchStatus.COMPLETED, execution.getStatus()); + assertEquals(2, execution.getStepExecutions().size()); + } + + @Test + public void testBuildOverTwoLines() throws Exception { + FlowJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step1).on("COMPLETED") + .to(step2).end(); + builder.preventRestart(); + builder.build().execute(execution); + assertEquals(BatchStatus.COMPLETED, execution.getStatus()); + assertEquals(2, execution.getStepExecutions().size()); + } + + @Test + public void testBuildSubflow() throws Exception { + Flow flow = new FlowBuilder("subflow").from(step1).end(); + JobFlowBuilder builder = new JobBuilder("flow").repository(jobRepository).start(flow); + builder.on("COMPLETED").to(step2); + builder.end().preventRestart().build().execute(execution); + assertEquals(BatchStatus.COMPLETED, execution.getStatus()); + assertEquals(2, execution.getStepExecutions().size()); + } + + @Test + public void testBuildSplit() throws Exception { + Flow flow = new FlowBuilder("subflow").from(step1).end(); + SimpleJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step2); + builder.split(new SimpleAsyncTaskExecutor()).add(flow).end(); + builder.preventRestart().build().execute(execution); + assertEquals(BatchStatus.COMPLETED, execution.getStatus()); + assertEquals(2, execution.getStepExecutions().size()); + } + + @Test + public void testBuildDecision() throws Exception { + JobExecutionDecider decider = new JobExecutionDecider() { + private int count = 0; + public FlowExecutionStatus decide(JobExecution jobExecution, StepExecution stepExecution) { + count++; + return count<2 ? new FlowExecutionStatus("ONGOING") : FlowExecutionStatus.COMPLETED; + } + }; + step1.setAllowStartIfComplete(true); + SimpleJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step1); + builder.next(decider).on("COMPLETED").end().from(decider).on("*").to(step1).end(); + builder.preventRestart().build().execute(execution); + assertEquals(BatchStatus.COMPLETED, execution.getStatus()); + assertEquals(2, execution.getStepExecutions().size()); + } + + @Test + public void testBuildWithIntermediateSimpleJob() throws Exception { + SimpleJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step1); + builder.on("COMPLETED").to(step2).end(); + builder.preventRestart(); + builder.build().execute(execution); + assertEquals(BatchStatus.COMPLETED, execution.getStatus()); + assertEquals(2, execution.getStepExecutions().size()); + } + + @Test + public void testBuildWithIntermediateSimpleJobTwoSteps() throws Exception { + SimpleJobBuilder builder = new JobBuilder("flow").repository(jobRepository).start(step1).next(step2); + builder.on("FAILED").to(step3).end(); + builder.build().execute(execution); + assertEquals(BatchStatus.COMPLETED, execution.getStatus()); + assertEquals(2, execution.getStepExecutions().size()); + } + +}