BATCH-679, BATCH-879: add Flow abstraction and FlowJob with support for decisions, splits and pauses.

spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/AbstractState.java

@@ -0,0 +1,40 @@
+/*
+ * 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.flow;
+
+
+/**
+ * @author Dave Syer
+ *
+ */
+public abstract class AbstractState<T> implements State<T> {
+	
+	private final String name;
+
+	/**
+	 * 
+	 */
+	public AbstractState(String name) {
+		this.name = name;
+	}
+
+	public String getName() {
+		return name;
+	}
+	
+	public abstract String handle(T context) throws Exception;
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/Flow.java

@@ -0,0 +1,44 @@
+/*
+ * 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.flow;
+
+
+/**
+ * @author Dave Syer
+ *
+ * @param <T>
+ */
+public interface Flow<T> {
+
+	/**
+	 * @return the name of the flow
+	 */
+	String getName();
+
+	/**
+	 * @throws FlowExecutionException
+	 */
+	FlowExecution start(T context) throws FlowExecutionException;
+
+	/**
+	 * @param stateName the name of the {@link State} to resume on
+	 * @param context the context to be passed into each {@link State} executed
+	 * @return a {@link FlowExecution} containing the exit status of the flow
+	 * @throws FlowExecutionException
+	 */
+	FlowExecution resume(String stateName, T context) throws FlowExecutionException;
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecution.java

@@ -0,0 +1,111 @@
+/*
+ * 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.flow;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class FlowExecution implements Comparable<FlowExecution> {
+
+	/**
+	 * Special well-known status value.
+	 */
+	public static final String COMPLETED = Status.COMPLETED.toString();
+
+	/**
+	 * Special well-known status value.
+	 */
+	public static final String PAUSED = Status.PAUSED.toString();
+	
+	/**
+	 * Special well-known status value.
+	 */
+	public static final String FAILED = Status.FAILED.toString();
+
+	/**
+	 * Special well-known status value.
+	 */
+	public static final String UNKNOWN = Status.UNKNOWN.toString();
+
+	private final String name;
+
+	private final String status;
+
+	private enum Status {
+
+		COMPLETED, PAUSED, FAILED, UNKNOWN;
+
+		static Status match(String value) {
+			for (int i = 0; i < values().length; i++) {
+				Status status = values()[i];
+				if (value.startsWith(status.toString())) {
+					return status;
+				}
+			}
+			// Default match should be the lowest priority
+			return COMPLETED;
+		}
+
+	};
+
+	/**
+	 * 
+	 */
+	public FlowExecution(String name, String status) {
+		this.name = name;
+		this.status = status;
+	}
+
+	/**
+	 * @return the name of the end state reached
+	 */
+	public String getName() {
+		return name;
+	}
+
+	/**
+	 * @return the exit status
+	 */
+	public String getStatus() {
+		return status;
+	}
+
+	/**
+	 * Create an ordering on {@link FlowExecution} instances by comparing their
+	 * statuses.
+	 * 
+	 * @see Comparable#compareTo(Object)
+	 * 
+	 * @param other
+	 * @return negative, zero or positive as per the contract
+	 */
+	public int compareTo(FlowExecution other) {
+		Status one = Status.match(this.getStatus());
+		Status two = Status.match(other.getStatus());
+		int comparison = one.compareTo(two);
+		if (comparison==0) {
+			return this.getStatus().compareTo(other.getStatus());
+		}
+		return comparison;
+	}
+
+	@Override
+	public String toString() {
+		return String.format("FlowExecution: name=%s, status=%s", name, status);
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionAggregator.java

@@ -0,0 +1,35 @@
+/*
+ * 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.flow;
+
+import java.util.Collection;
+
+/**
+ * Strategy interface for aggregating {@link FlowExecution} instances into a
+ * single exit status.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public interface FlowExecutionAggregator {
+
+	/**
+	 * @param executions the executions to aggregate
+	 * @return a summary status for the whole lot
+	 */
+	String aggregate(Collection<FlowExecution> executions);
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/FlowExecutionException.java

@@ -0,0 +1,39 @@
+/*
+ * 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.flow;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class FlowExecutionException extends Exception {
+
+	/**
+	 * @param message
+	 */
+	public FlowExecutionException(String message) {
+		super(message);
+	}
+
+	/**
+	 * @param message
+	 * @param cause
+	 */
+	public FlowExecutionException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/MaxValueFlowExecutionAggregator.java

@@ -0,0 +1,37 @@
+/*
+ * 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.flow;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class MaxValueFlowExecutionAggregator implements FlowExecutionAggregator {
+
+	/**
+	 * @see FlowExecutionAggregator#aggregate(Collection)
+	 */
+	public String aggregate(Collection<FlowExecution> executions) {
+		if (executions==null || executions.size()==0) {
+			return FlowExecution.UNKNOWN;
+		}
+		return Collections.max(executions).getStatus();
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/PatternMatcher.java

@@ -0,0 +1,162 @@
+/*
+ * 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.flow;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class PatternMatcher {
+
+	/**
+	 * Lifted from AntPathMatcher in Spring Core. Tests whether or not a string
+	 * matches against a pattern. The pattern may contain two special
+	 * characters:<br>
+	 * '*' means zero or more characters<br>
+	 * '?' means one and only one character
+	 * @param pattern pattern to match against. Must not be <code>null</code>.
+	 * @param str string which must be matched against the pattern. Must not be
+	 * <code>null</code>.
+	 * @return <code>true</code> if the string matches against the pattern, or
+	 * <code>false</code> otherwise.
+	 */
+	public static boolean match(String pattern, String str) {
+		char[] patArr = pattern.toCharArray();
+		char[] strArr = str.toCharArray();
+		int patIdxStart = 0;
+		int patIdxEnd = patArr.length - 1;
+		int strIdxStart = 0;
+		int strIdxEnd = strArr.length - 1;
+		char ch;
+
+		boolean containsStar = pattern.contains("*");
+
+		if (!containsStar) {
+			// No '*'s, so we make a shortcut
+			if (patIdxEnd != strIdxEnd) {
+				return false; // Pattern and string do not have the same size
+			}
+			for (int i = 0; i <= patIdxEnd; i++) {
+				ch = patArr[i];
+				if (ch != '?') {
+					if (ch != strArr[i]) {
+						return false;// Character mismatch
+					}
+				}
+			}
+			return true; // String matches against pattern
+		}
+
+		if (patIdxEnd == 0) {
+			return true; // Pattern contains only '*', which matches anything
+		}
+
+		// Process characters before first star
+		while ((ch = patArr[patIdxStart]) != '*' && strIdxStart <= strIdxEnd) {
+			if (ch != '?') {
+				if (ch != strArr[strIdxStart]) {
+					return false;// Character mismatch
+				}
+			}
+			patIdxStart++;
+			strIdxStart++;
+		}
+		if (strIdxStart > strIdxEnd) {
+			// All characters in the string are used. Check if only '*'s are
+			// left in the pattern. If so, we succeeded. Otherwise failure.
+			for (int i = patIdxStart; i <= patIdxEnd; i++) {
+				if (patArr[i] != '*') {
+					return false;
+				}
+			}
+			return true;
+		}
+
+		// Process characters after last star
+		while ((ch = patArr[patIdxEnd]) != '*' && strIdxStart <= strIdxEnd) {
+			if (ch != '?') {
+				if (ch != strArr[strIdxEnd]) {
+					return false;// Character mismatch
+				}
+			}
+			patIdxEnd--;
+			strIdxEnd--;
+		}
+		if (strIdxStart > strIdxEnd) {
+			// All characters in the string are used. Check if only '*'s are
+			// left in the pattern. If so, we succeeded. Otherwise failure.
+			for (int i = patIdxStart; i <= patIdxEnd; i++) {
+				if (patArr[i] != '*') {
+					return false;
+				}
+			}
+			return true;
+		}
+
+		// process pattern between stars. padIdxStart and patIdxEnd point
+		// always to a '*'.
+		while (patIdxStart != patIdxEnd && strIdxStart <= strIdxEnd) {
+			int patIdxTmp = -1;
+			for (int i = patIdxStart + 1; i <= patIdxEnd; i++) {
+				if (patArr[i] == '*') {
+					patIdxTmp = i;
+					break;
+				}
+			}
+			if (patIdxTmp == patIdxStart + 1) {
+				// Two stars next to each other, skip the first one.
+				patIdxStart++;
+				continue;
+			}
+			// Find the pattern between padIdxStart & padIdxTmp in str between
+			// strIdxStart & strIdxEnd
+			int patLength = (patIdxTmp - patIdxStart - 1);
+			int strLength = (strIdxEnd - strIdxStart + 1);
+			int foundIdx = -1;
+			strLoop: for (int i = 0; i <= strLength - patLength; i++) {
+				for (int j = 0; j < patLength; j++) {
+					ch = patArr[patIdxStart + j + 1];
+					if (ch != '?') {
+						if (ch != strArr[strIdxStart + i + j]) {
+							continue strLoop;
+						}
+					}
+				}
+
+				foundIdx = strIdxStart + i;
+				break;
+			}
+
+			if (foundIdx == -1) {
+				return false;
+			}
+
+			patIdxStart = patIdxTmp;
+			strIdxStart = foundIdx + patLength;
+		}
+
+		// All characters in the string are used. Check if only '*'s are left
+		// in the pattern. If so, we succeeded. Otherwise failure.
+		for (int i = patIdxStart; i <= patIdxEnd; i++) {
+			if (patArr[i] != '*') {
+				return false;
+			}
+		}
+
+		return true;
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SimpleFlow.java

@@ -0,0 +1,265 @@
+/*
+ * 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.flow;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+import org.springframework.beans.factory.InitializingBean;
+
+import com.sun.org.apache.xerces.internal.impl.xpath.XPath.Step;
+
+/**
+ * A {@link Flow} that branches conditionally depending on the exit status of
+ * the last {@link State}. The input parameters are the state transitions (in no
+ * particular order). The start state name can be specified explicitly (and must
+ * exist in the set of transitions), or computed from the existing transitions,
+ * if unambiguous.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class SimpleFlow<T> implements Flow<T> {
+
+	private State<T> startState;
+
+	private Map<String, SortedSet<StateTransition<T>>> transitionMap = new HashMap<String, SortedSet<StateTransition<T>>>();
+
+	private Map<String, State<T>> stateMap = new HashMap<String, State<T>>();
+
+	private String startStateName;
+
+	private Collection<StateTransition<T>> stateTransitions = new HashSet<StateTransition<T>>();
+
+	private final String name;
+
+	/**
+	 * Create a flow with the given name.
+	 * 
+	 * @param name the name of the flow
+	 */
+	public SimpleFlow(String name) {
+		this.name = name;
+	}
+
+	/**
+	 * Get the name for this flow.
+	 * 
+	 * @see Flow#getName()
+	 */
+	public String getName() {
+		return name;
+	}
+
+	/**
+	 * Public setter for the start state name.
+	 * @param startStateName the name of the start state
+	 */
+	public void setStartStateName(String startStateName) {
+		this.startStateName = startStateName;
+	}
+
+	/**
+	 * Public setter for the stateTransitions.
+	 * @param stateTransitions the stateTransitions to set
+	 */
+	public void setStateTransitions(Collection<StateTransition<T>> stateTransitions) {
+
+		this.stateTransitions = stateTransitions;
+	}
+
+	/**
+	 * Locate start step and pre-populate data structures needed for execution.
+	 * 
+	 * @see InitializingBean#afterPropertiesSet()
+	 */
+	public void afterPropertiesSet() throws Exception {
+		initializeTransitions();
+	}
+
+	/**
+	 * @see Flow#start(Object)
+	 */
+	public FlowExecution start(T context) throws FlowExecutionException {
+		if (startState == null) {
+			initializeTransitions();
+		}
+		State<T> state = startState;
+		String stateName = state.getName();
+		return resume(stateName, context);
+	}
+
+	/**
+	 * @see Flow#resume(String, Object)
+	 */
+	public FlowExecution resume(String stateName, T context) throws FlowExecutionException {
+		String status = FlowExecution.UNKNOWN;
+		State<T> state = stateMap.get(stateName);
+		// Terminate if there are no more states
+		while (state != null) {
+			stateName = state.getName();
+			try {
+				status = state.handle(context);
+			}
+			catch (Exception e) {
+				throw new FlowExecutionException(String.format("Ended flow=%s at state=%s with exception", name,
+						stateName), e);
+			}
+			state = nextState(stateName, status);
+		}
+		return new FlowExecution(stateName, status);
+	}
+
+	/**
+	 * @return the next {@link Step} (or null if this is the end)
+	 * @throws JobExecutionException
+	 */
+	private State<T> nextState(String stepName, String status) throws FlowExecutionException {
+
+		// Special status value indicating that a state wishes to pause
+		// execution
+		if (status.equals(FlowExecution.PAUSED)) {
+			return null;
+		}
+
+		Set<StateTransition<T>> set = transitionMap.get(stepName);
+
+		if (set == null) {
+			throw new FlowExecutionException(String.format("No transitions found in flow=%s for state=%s", getName(),
+					stepName));
+		}
+
+		String next = null;
+		for (StateTransition<T> stateTransition : set) {
+			if (stateTransition.matches(status)) {
+				if (stateTransition.isEnd()) {
+					// End of job
+					return null;
+				}
+				next = stateTransition.getNext();
+				break;
+			}
+		}
+
+		if (next == null) {
+			throw new FlowExecutionException(String.format(
+					"Next state not found in flow=%s for step=%s with exit status=%s", getName(), stepName, status));
+		}
+
+		if (!stateMap.containsKey(next)) {
+			throw new FlowExecutionException(String.format("Next state not specified in flow=%s for next=%s",
+					getName(), next));
+		}
+
+		return stateMap.get(next);
+
+	}
+
+	/**
+	 * Analyse the transitions provided and generate all the information needed
+	 * to execute the flow.
+	 */
+	private void initializeTransitions() {
+		startState = null;
+		transitionMap.clear();
+		stateMap.clear();
+		boolean hasEndStep = false;
+
+		for (StateTransition<T> stepTransition : stateTransitions) {
+			State<T> step = stepTransition.getState();
+			stateMap.put(step.getName(), step);
+		}
+
+		for (StateTransition<T> stateTransition : stateTransitions) {
+
+			State<T> state = stateTransition.getState();
+
+			if (!stateTransition.isEnd()) {
+
+				String next = stateTransition.getNext();
+
+				if (!stateMap.containsKey(next)) {
+					throw new IllegalArgumentException("Missing step for [" + stateTransition + "]");
+				}
+
+			}
+			else {
+				hasEndStep = true;
+			}
+
+			String name = state.getName();
+
+			SortedSet<StateTransition<T>> set = transitionMap.get(name);
+			if (set == null) {
+				set = new TreeSet<StateTransition<T>>();
+				transitionMap.put(name, set);
+			}
+			set.add(stateTransition);
+
+		}
+
+		if (!hasEndStep) {
+			throw new IllegalArgumentException(
+					"No end step was found.  You must specify at least one transition with no next state.");
+		}
+
+		if (startStateName != null) {
+
+			startState = stateMap.get(startStateName);
+			if (startState == null) {
+				throw new IllegalArgumentException(
+						"Start state does not exist (if you specify a startStateName make sure "
+								+ "a state with that name is in one of the transitions): [" + startStateName + "]");
+			}
+
+		}
+		else {
+
+			// Try and locate a transition with no incoming links
+
+			Set<String> nextStateNames = new HashSet<String>();
+
+			for (StateTransition<T> stepTransition : stateTransitions) {
+				nextStateNames.add(stepTransition.getNext());
+			}
+
+			for (StateTransition<T> stepTransition : stateTransitions) {
+				State<T> state = stepTransition.getState();
+				if (!nextStateNames.contains(state.getName())) {
+					if (startState != null && !startState.getName().equals(state.getName())) {
+						throw new IllegalArgumentException(String.format(
+								"Multiple possible start steps found: [%s, %s].  "
+										+ "Please specify one explicitly with the startStateName property.", startState
+										.getName(), state.getName()));
+					}
+					startState = state;
+				}
+			}
+
+			if (startState == null) {
+				throw new IllegalArgumentException(
+						"No start state could be located (no transition without incoming links)");
+			}
+
+		}
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/SplitState.java

@@ -0,0 +1,103 @@
+/*
+ * 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.flow;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.concurrent.Callable;
+import java.util.concurrent.FutureTask;
+
+import org.springframework.core.task.SyncTaskExecutor;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.core.task.TaskRejectedException;
+
+/**
+ * A {@link State} implementation that splits a {@link Flow} into multiple
+ * parallel subflows.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class SplitState<T> extends AbstractState<T> {
+
+	private final Collection<Flow<T>> flows;
+
+	private TaskExecutor taskExecutor = new SyncTaskExecutor();
+
+	private FlowExecutionAggregator aggregator = new MaxValueFlowExecutionAggregator();
+
+	/**
+	 * @param name
+	 */
+	public SplitState(String name, Collection<Flow<T>> flows) {
+		super(name);
+		this.flows = flows;
+	}
+
+	/**
+	 * Public setter for the taskExecutor.
+	 * @param taskExecutor the taskExecutor to set
+	 */
+	public void setTaskExecutor(TaskExecutor taskExecutor) {
+		this.taskExecutor = taskExecutor;
+	}
+
+	/**
+	 * Execute the flows in parallel by passing them to the {@link TaskExecutor}
+	 * and wait for all of them to finish before proceeding.
+	 * 
+	 * @see State#handle(Object)
+	 */
+	@Override
+	public String handle(final T context) throws Exception {
+
+		Collection<FutureTask<FlowExecution>> tasks = new ArrayList<FutureTask<FlowExecution>>();
+
+		for (final Flow<T> flow : flows) {
+
+			final FutureTask<FlowExecution> task = new FutureTask<FlowExecution>(new Callable<FlowExecution>() {
+				public FlowExecution call() throws Exception {
+					return flow.start(context);
+				}
+			});
+
+			tasks.add(task);
+
+			try {
+				taskExecutor.execute(new Runnable() {
+					public void run() {
+						task.run();
+					}
+				});
+			}
+			catch (TaskRejectedException e) {
+				throw new FlowExecutionException("TaskExecutor rejected task for flow=" + flow.getName());
+			}
+
+		}
+
+		Collection<FlowExecution> results = new ArrayList<FlowExecution>();
+
+		// TODO: could use a CompletionSerice?
+		for (FutureTask<FlowExecution> task : tasks) {
+			results.add(task.get());
+		}
+
+		return aggregator.aggregate(results);
+
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/State.java

@@ -0,0 +1,47 @@
+/*
+ * 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.flow;
+
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public interface State<T> {
+
+	/**
+	 * The name of the state. Should be unique within a flow.
+	 * 
+	 * @return the name of this state
+	 */
+	String getName();
+
+	/**
+	 * Handle some business or processing logic and return a status that can be
+	 * used to drive a flow to the next {@link State}. The status can be any
+	 * string, but special meaning is assigned to the static constants in
+	 * {@link FlowExecution}. The context can be used by implementations to do
+	 * whatever they need to do. The same context will be passed to all
+	 * {@link State} instances, so implementations should be careful that the
+	 * context is thread safe, or used in a thread safe manner.
+	 * 
+	 * @param context the context passed in by the caller
+	 * @return a status for the execution
+	 * @throws Exception if anything goes wrong
+	 */
+	String handle(T context) throws Exception;
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/flow/StateTransition.java

@@ -0,0 +1,180 @@
+/*
+ * 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.flow;
+
+import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.util.StringUtils;
+
+/**
+ * Value object representing a potential transition from one {@link State} to
+ * another. The originating State name and the next {@link State} to execute are
+ * linked by a pattern for the {@link ExitStatus#getExitCode() exit code} of an
+ * execution of the originating State.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class StateTransition<T> implements Comparable<StateTransition<T>> {
+
+	private final State<T> state;
+
+	private final String pattern;
+
+	private final String next;
+
+	/**
+	 * Create a new end state {@link StateTransition} specification. This
+	 * transition explicitly goes unconditionally to an end state (i.e. no more
+	 * executions).
+	 * 
+	 * @param state the {@link State} used to generate the outcome for this
+	 * transition
+	 */
+	public static <T> StateTransition<T> createEndStateTransition(State<T> state) {
+		return createStateTransition(state, null, null);
+	}
+
+	/**
+	 * Create a new end state {@link StateTransition} specification. This
+	 * transition explicitly goes to an end state (i.e. no more processing) if
+	 * the outcome matches the pattern.
+	 * 
+	 * @param state the {@link State} used to generate the outcome for this
+	 * transition
+	 * @param pattern the pattern to match in the exit status of the
+	 * {@link State}
+	 */
+	public static <T> StateTransition<T> createEndStateTransition(State<T> state, String pattern) {
+		return createStateTransition(state, pattern, null);
+	}
+
+	/**
+	 * Create a new state {@link StateTransition} specification with a wildcard
+	 * pattern that matches all outcomes.
+	 * 
+	 * @param state the {@link State} used to generate the outcome for this
+	 * transition
+	 * @param next the name of the next {@link State} to execute
+	 */
+	public static <T> StateTransition<T> createStateTransition(State<T> state, String next) {
+		return createStateTransition(state, null, next);
+	}
+
+	/**
+	 * Create a new {@link StateTransition} specification from one {@link State}
+	 * to another (by name).
+	 * 
+	 * @param state the {@link State} used to generate the outcome for this
+	 * transition
+	 * @param pattern the pattern to match in the exit status of the
+	 * {@link State}
+	 * @param next the name of the next {@link State} to execute
+	 */
+	public static <T> StateTransition<T> createStateTransition(State<T> state, String pattern, String next) {
+		return new StateTransition<T>(state, pattern, next);
+	}
+
+	private StateTransition(State<T> state, String pattern, String next) {
+		super();
+		if (!StringUtils.hasText(pattern)) {
+			this.pattern = "*";
+		}
+		else {
+			this.pattern = pattern;
+		}
+		this.next = next;
+		this.state = state;
+	}
+
+	/**
+	 * Public getter for the State.
+	 * @return the State
+	 */
+	public State<T> getState() {
+		return state;
+	}
+
+	/**
+	 * Public getter for the next State name.
+	 * @return the next
+	 */
+	public String getNext() {
+		return next;
+	}
+
+	/**
+	 * Check if the provided status matches the pattern, signalling that the
+	 * next State should be executed.
+	 * 
+	 * @param status the status to compare
+	 * @return true if the pattern matches this status
+	 */
+	public boolean matches(String status) {
+		return PatternMatcher.match(pattern, status);
+	}
+
+	/**
+	 * Check for a special next State signalling the end of a job.
+	 * 
+	 * @return true if this transition goes nowhere (there is no next)
+	 */
+	public boolean isEnd() {
+		return next == null;
+	}
+
+	/**
+	 * Sorts by decreasing specificity of pattern, based on just counting
+	 * wildcards (with * taking precedence over ?). If wildcard counts are equal
+	 * then falls back to alphabetic comparison. Hence * &gt; foo* &gt; ??? &gt;
+	 * fo? > foo.
+	 * @see Comparable#compareTo(Object)
+	 */
+	public int compareTo(StateTransition<T> other) {
+		String value = other.pattern;
+		if (pattern.equals(value)) {
+			return 0;
+		}
+		int patternCount = StringUtils.countOccurrencesOf(pattern, "*");
+		int valueCount = StringUtils.countOccurrencesOf(value, "*");
+		if (patternCount > valueCount) {
+			return 1;
+		}
+		if (patternCount < valueCount) {
+			return -1;
+		}
+		patternCount = StringUtils.countOccurrencesOf(pattern, "?");
+		valueCount = StringUtils.countOccurrencesOf(value, "?");
+		if (patternCount > valueCount) {
+			return 1;
+		}
+		if (patternCount < valueCount) {
+			return -1;
+		}
+		return pattern.compareTo(value);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see java.lang.Object#toString()
+	 */
+	@Override
+	public String toString() {
+		return String.format("StateTransition: state=%s, pattern=%s, next=%s", state == null ? null : state.getName(),
+				pattern, next);
+	}
+
+}