Add Spring Integration prototype code

spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/AsynchronousFailureException.java

@@ -0,0 +1,37 @@
+package org.springframework.integration.batch.chunk;
+
+import org.springframework.batch.item.ItemWriterException;
+
+/**
+ * Exception indicating that a failure or early completion condition was
+ * detected in a remote worker.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class AsynchronousFailureException extends ItemWriterException {
+
+	/**
+	 * Create a new {@link AsynchronousFailureException} based on a message and
+	 * another exception.
+	 * 
+	 * @param message
+	 *            the message for this exception
+	 * @param cause
+	 *            the other exception
+	 */
+	public AsynchronousFailureException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+	/**
+	 * Create a new {@link AsynchronousFailureException} based on a message.
+	 * 
+	 * @param message
+	 *            the message for this exception
+	 */
+	public AsynchronousFailureException(String message) {
+		super(message);
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkHandler.java

@@ -0,0 +1,8 @@
+package org.springframework.integration.batch.chunk;
+
+
+public interface ChunkHandler {
+
+	ChunkResponse handleChunk(ChunkRequest chunk);
+
+}

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

@@ -0,0 +1,255 @@
+package org.springframework.integration.batch.chunk;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.listener.StepExecutionListenerSupport;
+import org.springframework.batch.item.ClearFailedException;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.FlushFailedException;
+import org.springframework.batch.item.ItemStream;
+import org.springframework.batch.item.ItemStreamException;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.item.database.HibernateAwareItemWriter;
+import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.batch.repeat.RepeatContext;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.message.Message;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.util.Assert;
+
+public class ChunkMessageChannelItemWriter extends StepExecutionListenerSupport implements ItemWriter, ItemStream {
+
+	private static final Log logger = LogFactory.getLog(ChunkMessageChannelItemWriter.class);
+
+	/**
+	 * Key for items processed in the current transaction {@link RepeatContext}.
+	 */
+	private static final String ITEMS_PROCESSED = HibernateAwareItemWriter.class.getName() + ".ITEMS_PROCESSED";
+
+	static final String ACTUAL = "ACTUAL";
+
+	static final String EXPECTED = "EXPECTED";
+
+	private static final long DEFAULT_THROTTLE_LIMIT = 6;
+
+	private MessageChannel requestChannel;
+
+	private MessageChannel replyChannel;
+
+	// TODO: abstract the state or make a factory for this writer?
+	private LocalState localState = new LocalState();
+
+	private long throttleLimit = DEFAULT_THROTTLE_LIMIT;
+
+	public void setReplyChannel(MessageChannel replyChannel) {
+		this.replyChannel = replyChannel;
+	}
+
+	public void setRequestChannel(MessageChannel requestChannel) {
+		this.requestChannel = requestChannel;
+	}
+
+	public void write(Object item) throws Exception {
+		bindTransactionResources();
+		getProcessed().add(item);
+		logger.debug("Added item to chunk: " + item);
+	}
+
+	/**
+	 * Flush the buffer.
+	 * 
+	 * @see org.springframework.batch.item.ItemWriter#flush()
+	 */
+	public void flush() throws FlushFailedException {
+
+		bindTransactionResources(); // in case we are called outside a
+		// transaction
+
+		// Block until expecting < throttle limit - can Spring
+		// Integration do that for me?
+		while (localState.getExpecting() > throttleLimit) {
+			getNextResult(100);
+		}
+
+		List<Object> processed = getProcessed();
+
+		if (!processed.isEmpty()) {
+
+			logger.debug("Dispatching chunk: " + processed);
+			ChunkRequest request = new ChunkRequest(processed, localState.getJobId(), localState.getSkipCount());
+			GenericMessage<ChunkRequest> message = new GenericMessage<ChunkRequest>(request );
+			requestChannel.send(message);
+			localState.expected++;
+
+		}
+
+		// Short little timeout to look for an immediate reply.
+		getNextResult(1);
+
+		unbindTransactionResources();
+
+	}
+	
+	@Override
+	public void beforeStep(StepExecution stepExecution) {
+		localState.setStepExecution(stepExecution);
+	}
+
+	@Override
+	public ExitStatus afterStep(StepExecution stepExecution) {
+		if (!(stepExecution.getStatus() == BatchStatus.COMPLETED)) {
+			return ExitStatus.CONTINUABLE;
+		}
+		long expecting = localState.getExpecting();
+		boolean timedOut;
+		try {
+			logger.debug("Waiting for results in step listener...");
+			timedOut = !waitForResults();
+			logger.debug("Finished waiting for results in step listener.");
+		}
+		catch (RuntimeException e) {
+			logger.debug("Detected failure waiting for results in step listener.");
+			stepExecution.setStatus(BatchStatus.FAILED);
+			return ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": " + e.getMessage());
+		}
+		if (timedOut) {
+			stepExecution.setStatus(BatchStatus.FAILED);
+			throw new ItemStreamException("Timed out waiting for back log at end of step");
+		}
+		return ExitStatus.FINISHED.addExitDescription("Waited for " + expecting + " results.");
+	}
+
+	public void close(ExecutionContext executionContext) throws ItemStreamException {
+		localState.reset();
+	}
+
+	public void open(ExecutionContext executionContext) throws ItemStreamException {
+		if (executionContext.containsKey(EXPECTED)) {
+			localState.expected = executionContext.getLong(EXPECTED);
+			localState.actual = executionContext.getLong(ACTUAL);
+			if (!waitForResults()) {
+				throw new ItemStreamException("Timed out waiting for back log on open");
+			}
+		}
+	}
+
+	public void update(ExecutionContext executionContext) throws ItemStreamException {
+		executionContext.putLong(EXPECTED, localState.expected);
+		executionContext.putLong(ACTUAL, localState.actual);
+	}
+
+	/**
+	 * Wait until all the results that are in the pipeline come back to the
+	 * reply channel.
+	 * 
+	 * @return true if successfully received a result, false if timed out
+	 */
+	private boolean waitForResults() {
+		// TODO: cumulative timeout, or throw an exception?
+		int count = 0;
+		while (localState.getExpecting() > 0 && count++ < 10) {
+			getNextResult(100);
+		}
+		return count < 10;
+	}
+
+	/**
+	 * Get the next result if it is available within the timeout specified,
+	 * otherwise return null.
+	 */
+	private void getNextResult(long timeout) {
+		Message<?> message = replyChannel.receive(timeout);
+		if (message != null) {
+			ChunkResponse payload = (ChunkResponse) message.getPayload();
+			Long jobInstanceId = payload.getJobId();
+			Assert.state(jobInstanceId!=null, "Message did not contain job instance id.");
+			Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id ["+jobInstanceId+"] should have been ["+localState.getJobId()+"].");
+			localState.actual++;
+			ExitStatus result = payload.getExitStatus();
+			// TODO: check it can never be ExitStatus.FINISHED?
+			if (!result.isContinuable()) {
+				throw new AsynchronousFailureException("Failure or early completion detected in handler: " + result);
+			}
+		}
+	}
+
+	/**
+	 * Accessor for the list of processed items in this transaction.
+	 * 
+	 * @return the processed
+	 */
+	@SuppressWarnings("unchecked")
+	private List<Object> getProcessed() {
+		Assert.state(TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED),
+				"Processed items not bound to transaction.");
+		List<Object> processed = (List<Object>) TransactionSynchronizationManager.getResource(ITEMS_PROCESSED);
+		return processed;
+	}
+
+	/**
+	 * Set up the {@link RepeatContext} as a transaction resource.
+	 * 
+	 * @param context the context to set
+	 */
+	private void bindTransactionResources() {
+		if (TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED)) {
+			return;
+		}
+		TransactionSynchronizationManager.bindResource(ITEMS_PROCESSED, new ArrayList<Object>());
+	}
+
+	/**
+	 * Remove the transaction resource associated with this context.
+	 */
+	private void unbindTransactionResources() {
+		if (!TransactionSynchronizationManager.hasResource(ITEMS_PROCESSED)) {
+			return;
+		}
+		TransactionSynchronizationManager.unbindResource(ITEMS_PROCESSED);
+	}
+
+	/**
+	 * Clear the buffer.
+	 * 
+	 * @see org.springframework.batch.item.ItemWriter#clear()
+	 */
+	public void clear() throws ClearFailedException {
+		unbindTransactionResources();
+	}
+
+	private static class LocalState {
+		private long actual;
+
+		private long expected;
+
+		private StepExecution stepExecution;
+
+		public long getExpecting() {
+			return expected - actual;
+		}
+
+		public int getSkipCount() {
+			// TODO Auto-generated method stub
+			return stepExecution.getSkipCount();
+		}
+
+		public Long getJobId() {
+			return stepExecution.getJobExecution().getJobId();
+		}
+
+		public void setStepExecution(StepExecution stepExecution) {
+			this.stepExecution = stepExecution;
+		}
+
+		public void reset() {
+			expected = actual = 0;
+		}
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkRequest.java

@@ -0,0 +1,30 @@
+package org.springframework.integration.batch.chunk;
+
+import java.io.Serializable;
+import java.util.Collection;
+
+public class ChunkRequest implements Serializable {
+
+	private final int skipCount;
+	private final Long jobId;
+	private final Collection<Object> items;
+
+	public ChunkRequest(Collection<Object> items, Long jobId, int skipCount) {
+		this.items = items;
+		this.jobId = jobId;
+		this.skipCount = skipCount;
+	}
+
+	public int getSkipCount() {
+		return skipCount;
+	}
+
+	public Long getJobId() {
+		return jobId;
+	}
+
+	public Collection<Object> getItems() {
+		return items;
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ChunkResponse.java

@@ -0,0 +1,31 @@
+package org.springframework.integration.batch.chunk;
+
+import java.io.Serializable;
+
+import org.springframework.batch.repeat.ExitStatus;
+
+public class ChunkResponse implements Serializable {
+
+	private final int skipCount;
+	private final Long jobId;
+	private final ExitStatus exitStatus;
+
+	public ChunkResponse(ExitStatus exitStatus, Long jobId, int skipCount) {
+		this.exitStatus = exitStatus;
+		this.jobId = jobId;
+		this.skipCount = skipCount;
+	}
+
+	public int getSkipCount() {
+		return skipCount;
+	}
+
+	public Long getJobId() {
+		return jobId;
+	}
+
+	public ExitStatus getExitStatus() {
+		return exitStatus;
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/chunk/ItemWriterChunkHandler.java

@@ -0,0 +1,85 @@
+package org.springframework.integration.batch.chunk;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.core.SkipListener;
+import org.springframework.batch.core.listener.CompositeSkipListener;
+import org.springframework.batch.core.step.skip.ItemSkipPolicy;
+import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.integration.annotation.Handler;
+import org.springframework.transaction.annotation.Transactional;
+
+public class ItemWriterChunkHandler implements ChunkHandler {
+
+	private static final Log logger = LogFactory.getLog(ItemWriterChunkHandler.class);
+
+	private ItemWriter itemWriter;
+
+	private ItemSkipPolicy itemSkipPolicy = new NeverSkipItemSkipPolicy();
+
+	private CompositeSkipListener skipListener = new CompositeSkipListener();
+
+	public void setItemSkipPolicy(ItemSkipPolicy itemSkipPolicy) {
+		this.itemSkipPolicy = itemSkipPolicy;
+	}
+
+	public void setItemWriter(ItemWriter itemWriter) {
+		this.itemWriter = itemWriter;
+	}
+
+	public void registerSkipListener(SkipListener listener) {
+		skipListener.register(listener);
+	}
+
+	public void setSkipListeners(SkipListener[] skipListeners) {
+		for (SkipListener listener : skipListeners) {
+			registerSkipListener(listener);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.integration.batch.slave.ChunkHandler#handleChunk(java.util.Collection)
+	 */
+	@Handler
+	@Transactional
+	public ChunkResponse handleChunk(ChunkRequest chunk) {
+
+		logger.debug("Handling chunk: " + chunk);
+
+		int parentSkipCount = chunk.getSkipCount();
+		int skipCount = 0;
+
+		try {
+			for (Object item : chunk.getItems()) {
+				try {
+					itemWriter.write(item);
+				}
+				catch (Exception e) {
+					if (itemSkipPolicy.shouldSkip(e, parentSkipCount + skipCount)) {
+						logger.debug("Skipping item on exception", e);
+						skipCount++;
+						skipListener.onSkipInWrite(item, e);
+					} else {
+						logger.debug("Cannot skip, re-throwing");
+						throw e;
+					}
+				}
+			}
+			itemWriter.flush();
+		}
+		catch (Exception e) {
+			logger.debug("Failed chunk", e);
+			itemWriter.clear();
+			// TODO: need to force rollback as well
+			return new ChunkResponse(ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": "
+					+ e.getMessage()), chunk.getJobId(), skipCount);
+		}
+
+		logger.debug("Completed chunk handling with " + skipCount + " skips");
+		return new ChunkResponse(ExitStatus.CONTINUABLE, chunk.getJobId(), skipCount);
+
+	}
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/file/FileToMessagesJobFactoryBean.java

@@ -0,0 +1,185 @@
+/*
+ * 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.integration.batch.file;
+
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.StepExecutionListener;
+import org.springframework.batch.core.job.SimpleJob;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.resource.StepExecutionResourceProxy;
+import org.springframework.batch.core.step.item.SimpleStepFactoryBean;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.file.FlatFileItemReader;
+import org.springframework.batch.item.xml.StaxEventItemReader;
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.beans.factory.annotation.Required;
+import org.springframework.core.io.Resource;
+import org.springframework.integration.batch.item.MessageChannelItemWriter;
+import org.springframework.integration.batch.launch.JobLaunchingMessageHandler;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.dispatcher.DirectChannel;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.util.Assert;
+
+/**
+ * A FactoryBean for a {@link Job} with a single step which just pumps messages
+ * from a file into a channel. The channel has to be a
+ * {@link SynchronousChannel} to ensure that failures propagate up to the step
+ * and fail the job execution. Normally this job will be used in conjunction
+ * with a {@link JobLaunchingMessageHandler} and a
+ * {@link ResourcePayloadAsJobParameterStrategy}, so that the user can just
+ * send a message to a request channel listing the files to be processed, and
+ * everything else just happens by magic. After a failure the job will be
+ * restarted just by sending it the same message.
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class FileToMessagesJobFactoryBean implements FactoryBean, BeanNameAware {
+
+	private String name = "fileToMessageJob";
+
+	private ItemReader itemReader;
+
+	private MessageChannel channel;
+
+	private PlatformTransactionManager transactionManager;
+
+	private JobRepository jobRepository;
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String)
+	 */
+	public void setBeanName(String name) {
+		this.name = name;
+	}
+
+	/**
+	 * Public setter for the {@link ItemReader}. Must be either a
+	 * {@link FlatFileItemReader} or a {@link StaxEventItemReader}. In either
+	 * case there is no need to set the resource property as it will be set by
+	 * this factory.
+	 * 
+	 * @param itemReader the itemReader to set
+	 */
+	@Required
+	public void setItemReader(ItemReader itemReader) {
+		this.itemReader = itemReader;
+	}
+
+	/**
+	 * Public setter for the channel. Each item from the item reader will be
+	 * sent to this channel.
+	 * @param channel the channel to set
+	 */
+	@Required
+	public void setChannel(MessageChannel channel) {
+		this.channel = channel;
+	}
+
+	/**
+	 * Public setter for the {@link JobRepository}.
+	 * @param jobRepository the job repository to set
+	 */
+	@Required
+	public void setJobRepository(JobRepository jobRepository) {
+		this.jobRepository = jobRepository;
+	}
+
+	/**
+	 * Public setter for the {@link PlatformTransactionManager}.
+	 * @param transactionManager the transaction manager to set
+	 */
+	@Required
+	public void setTransactionManager(PlatformTransactionManager transactionManager) {
+		this.transactionManager = transactionManager;
+	}
+
+	/**
+	 * Creates a {@link Job} that can process a flat file or XML file into
+	 * messages. To launch the job will require only a {@link JobParameters}
+	 * instance with a resource location as a URL.
+	 * 
+	 * @see org.springframework.beans.factory.FactoryBean#getObject()
+	 */
+	public Object getObject() throws Exception {
+
+		SimpleJob job = new SimpleJob();
+		job.setName(name);
+		job.setJobRepository(jobRepository);
+
+		SimpleStepFactoryBean stepFactory = new SimpleStepFactoryBean();
+		stepFactory.setBeanName("step");
+
+		Assert.state((itemReader instanceof FlatFileItemReader) || (itemReader instanceof StaxEventItemReader),
+				"ItemReader must be either a FlatFileItemReader or a StaxEventItemReader");
+		StepExecutionResourceProxy resourceProxy = new StepExecutionResourceProxy();
+		resourceProxy.setFilePattern("%" + ResourcePayloadAsJobParameterStrategy.FILE_INPUT_PATH + "%");
+		stepFactory.setListeners(new StepExecutionListener[] { resourceProxy });
+		setResource(itemReader, resourceProxy);
+		stepFactory.setItemReader(itemReader);
+
+		Assert.notNull(channel, "A channel must be provided");
+		Assert.state(channel instanceof DirectChannel,
+				"The channel must be a DirectChannel (otherwise failures can not be recovered from)");
+		MessageChannelItemWriter itemWriter = new MessageChannelItemWriter();
+		itemWriter.setChannel(channel);
+		stepFactory.setItemWriter(itemWriter);
+
+		Assert.notNull(transactionManager, "A transaction manager must be provided");
+		stepFactory.setTransactionManager(transactionManager);
+
+		Assert.notNull(jobRepository, "A job repository must be provided");
+		stepFactory.setJobRepository(jobRepository);
+
+		job.addStep((Step) stepFactory.getObject());
+		return job;
+	}
+
+	/**
+	 * @param itemReader
+	 * @param resource
+	 */
+	private void setResource(ItemReader itemReader, Resource resource) {
+		if (itemReader instanceof FlatFileItemReader) {
+			((FlatFileItemReader) itemReader).setResource(resource);
+		}
+		else {
+			((StaxEventItemReader) itemReader).setResource(resource);
+		}
+	}
+
+	/**
+	 * Always returns {@link Job}.
+	 * @see org.springframework.beans.factory.FactoryBean#getObjectType()
+	 */
+	public Class<?> getObjectType() {
+		return Job.class;
+	}
+
+	/**
+	 * Always true. TODO: should it be false?
+	 * @see org.springframework.beans.factory.FactoryBean#isSingleton()
+	 */
+	public boolean isSingleton() {
+		return true;
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/file/ResourcePayloadAsJobParameterStrategy.java

@@ -0,0 +1,56 @@
+/*
+ * 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.integration.batch.file;
+
+import java.io.IOException;
+
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.batch.item.ItemStreamException;
+import org.springframework.core.io.Resource;
+import org.springframework.integration.batch.launch.MessageToJobParametersStrategy;
+import org.springframework.integration.message.Message;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class ResourcePayloadAsJobParameterStrategy implements MessageToJobParametersStrategy {
+
+	/**
+	 * The key name for the job parameter that will be a URL for the input file
+	 */
+	public static final String FILE_INPUT_PATH = "input.file.path";
+
+	/**
+	 * Convert a message payload which is a {@link Resource} to its URL
+	 * representation and load that into a job parameter.
+	 * 
+	 * @see org.springframework.integration.batch.launch.MessageToJobParametersStrategy#getJobParameters(org.springframework.integration.message.Message)
+	 */
+	public JobParameters getJobParameters(Message<?> message) {
+		JobParametersBuilder builder = new JobParametersBuilder();
+		Resource resource = (Resource) message.getPayload();
+		try {
+			builder.addString(FILE_INPUT_PATH, resource.getURL().toExternalForm());
+		}
+		catch (IOException e) {
+			throw new ItemStreamException("Could not create URL for resource: [" + resource + "]", e);
+		}
+		return builder.toJobParameters();
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/item/MessageChannelItemWriter.java

@@ -0,0 +1,48 @@
+/*
+ * 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.integration.batch.item;
+
+import org.springframework.batch.item.AbstractItemWriter;
+import org.springframework.beans.factory.annotation.Required;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.GenericMessage;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class MessageChannelItemWriter extends AbstractItemWriter {
+
+	private MessageChannel channel;
+
+	/**
+	 * Public setter for the channel.
+	 * @param channel the channel to set
+	 */
+	@Required
+	public void setChannel(MessageChannel channel) {
+		this.channel = channel;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.batch.item.ItemWriter#write(java.lang.Object)
+	 */
+	public void write(Object item) throws Exception {
+		channel.send(new GenericMessage<Object>(item));
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/job/JobExecutionReply.java

@@ -0,0 +1,24 @@
+/*
+ * 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.integration.batch.job;
+
+/**
+ * @author dsyer
+ *
+ */
+public class JobExecutionReply {
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/job/JobExecutionRequest.java

@@ -0,0 +1,107 @@
+/*
+ * 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.integration.batch.job;
+
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.core.AttributeAccessorSupport;
+
+/**
+ * Encapsulation of a request to execute a job execution through a message flow
+ * consisting of step handlers. A handler should pass the message on as it is,
+ * modifying the request properties as necessary. Generally a handler will
+ * execute a step as part of the {@link JobExecution} passed in, and should
+ * change the status to {@link BatchStatus#COMPLETED} if the step is successful
+ * (generally a handler cannot determine if the whole job execution is complete,
+ * so this is just information about the step).<br/>
+ * 
+ * If the incoming status is {@link BatchStatus#FAILED},
+ * {@link BatchStatus#STOPPED} or {@link BatchStatus#STOPPING} the request
+ * should be ignored by handlers (passed on without modification).
+ * 
+ * @author Dave Syer
+ * 
+ */
+public class JobExecutionRequest extends AttributeAccessorSupport {
+
+	private JobExecution jobExecution;
+
+	private BatchStatus status;
+
+	private Throwable throwable;
+
+	/**
+	 * @param jobExecution
+	 */
+	public JobExecutionRequest(JobExecution jobExecution) {
+		this.jobExecution = jobExecution;
+		status = jobExecution.getStatus();
+	}
+
+	/**
+	 * @return
+	 */
+	public Long getJobId() {
+		return this.jobExecution.getJobId();
+	}
+
+	/**
+	 * @return
+	 */
+	public BatchStatus getStatus() {
+		return status;
+	}
+
+	/**
+	 * Public setter for the status.
+	 * @param status the status to set
+	 */
+	public void setStatus(BatchStatus status) {
+		this.status = status;
+	}
+
+	/**
+	 * @return
+	 */
+	public boolean hasErrors() {
+		return throwable != null;
+	}
+
+	/**
+	 * Public getter for the throwable.
+	 * @return the throwable
+	 */
+	public Throwable getLastThrowable() {
+		return throwable;
+	}
+
+	/**
+	 * Public setter for the throwable.
+	 * @param throwable the throwable to set
+	 */
+	public void registerThrowable(Throwable throwable) {
+		this.throwable = throwable;
+	}
+
+	/**
+	 * Public getter for the jobExecution.
+	 * @return the jobExecution
+	 */
+	public JobExecution getJobExecution() {
+		return jobExecution;
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/job/MessageOrientedStep.java

@@ -0,0 +1,165 @@
+/*
+ * 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.integration.batch.job;
+
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.JobInterruptedException;
+import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.core.UnexpectedJobExecutionException;
+import org.springframework.batch.core.step.AbstractStep;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.beans.factory.annotation.Required;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.message.Message;
+import org.springframework.util.Assert;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class MessageOrientedStep extends AbstractStep {
+
+	/**
+	 * Key in execution context for flag to say we are waiting.
+	 */
+	public static final String WAITING = MessageOrientedStep.class.getName() + ".WAITING";
+
+	private MessageChannel requestChannel;
+
+	private MessageChannel replyChannel;
+
+	/**
+	 * Public setter for the requestChannel.
+	 * @param requestChannel the requestChannel to set
+	 */
+	@Required
+	public void setRequestChannel(MessageChannel requestChannel) {
+		this.requestChannel = requestChannel;
+	}
+
+	/**
+	 * Public setter for the replyChannel.
+	 * @param replyChannel the replyChannel to set
+	 */
+	@Required
+	public void setReplyChannel(MessageChannel replyChannel) {
+		this.replyChannel = replyChannel;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.batch.core.step.AbstractStep#execute(org.springframework.batch.core.StepExecution)
+	 */
+	@Override
+	public ExitStatus doExecute(StepExecution stepExecution) throws JobInterruptedException,
+			UnexpectedJobExecutionException {
+
+		JobExecutionRequest request = new JobExecutionRequest(stepExecution.getJobExecution());
+
+		ExecutionContext executionContext = stepExecution.getExecutionContext();
+
+		if (executionContext.containsKey(WAITING)) {
+			// restart scenario: we are still waiting for a response
+			waitForReply(request.getJobId());
+		}
+		else {
+			executionContext.putString(WAITING, "true");
+			// TODO: need these two lines to be atomic
+			getJobRepository().saveOrUpdate(stepExecution);
+			requestChannel.send(new GenericMessage<JobExecutionRequest>(request));
+			waitForReply(request.getJobId());
+		}
+
+		return ExitStatus.FINISHED;
+
+	}
+	
+	/**
+	 * Do nothing.
+	 * 
+	 * @see org.springframework.batch.core.step.AbstractStep#open(org.springframework.batch.item.ExecutionContext)
+	 */
+	@Override
+	protected void open(ExecutionContext ctx) throws Exception {
+	}
+	
+	/**
+	 * Do nothing.
+	 * 
+	 * @see org.springframework.batch.core.step.AbstractStep#close(org.springframework.batch.item.ExecutionContext)
+	 */
+	@Override
+	protected void close(ExecutionContext ctx) throws Exception {
+	}
+
+	/**
+	 * @param expectedJobId
+	 */
+	private void waitForReply(Long expectedJobId) {
+		// TODO: promote timeout to field and calculate count
+		long timeout = 5;
+		int count = 0;
+
+		// TODO: use a ReponseCorrelator?, or just a SynchronousChannel
+		while (count++ < 100) {
+
+			Message<?> message = replyChannel.receive(timeout);
+
+			if (message != null) {
+
+				JobExecutionRequest payload = (JobExecutionRequest) message.getPayload();
+				Long jobInstanceId = payload.getJobId();
+				Assert.state(jobInstanceId != null, "Message did not contain job instance id.");
+				Assert.state(jobInstanceId.equals(expectedJobId), "Message contained wrong job instance id ["
+						+ jobInstanceId + "] should have been [" + expectedJobId + "].");
+
+				if (payload.getStatus() == BatchStatus.COMPLETED) {
+					// One of the steps decided we were finished
+					// TODO: wait for all the other steps that might be
+					// executing concurrently?
+					// TODO: maybe *any* reply on this channel should
+					// mean the end of the step?
+					break;
+				}
+
+				if (payload.hasErrors()) {
+					rethrow(payload.getLastThrowable());
+				}
+
+			}
+		}
+
+		if (count >= 100) {
+			throw new StepExecutionTimeoutException("Timed out waiting for steps to execute.");
+		}
+	}
+
+	/**
+	 * @param lastThrowable
+	 */
+	private static void rethrow(Throwable t) throws RuntimeException {
+		if (t instanceof RuntimeException) {
+			throw (RuntimeException) t;
+		}
+		if (t instanceof Exception) {
+			throw new UnexpectedJobExecutionException("Unexpected checked exception thrown by step.", t);
+		}
+		throw (Error) t;
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/job/StepExecutionMessageHandler.java

@@ -0,0 +1,238 @@
+/*
+ * 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.integration.batch.job;
+
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobExecutionException;
+import org.springframework.batch.core.JobInstance;
+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.item.ExecutionContext;
+import org.springframework.batch.repeat.ExitStatus;
+import org.springframework.beans.factory.annotation.Required;
+import org.springframework.integration.annotation.Handler;
+
+/**
+ * @author Dave Syer
+ * 
+ */
+public class StepExecutionMessageHandler {
+
+	private Step step;
+
+	private JobRepository jobRepository;
+
+	private String[] inputKeys = new String[0];
+
+	private String[] outputKeys = new String[0];
+
+	/**
+	 * Public setter for the {@link Step}.
+	 * @param step the step to set
+	 */
+	@Required
+	public void setStep(Step step) {
+		this.step = step;
+	}
+
+	/**
+	 * Public setter for the input keys. Attributes from the incoming request
+	 * will be added to the execution context for the step.
+	 * @param inputKeys the inputKeys to set
+	 */
+	public void setInputKeys(String[] inputKeys) {
+		this.inputKeys = inputKeys;
+	}
+
+	/**
+	 * Public setter for the output keys. Attributes from a successful step
+	 * execution context will be added to the request before it is handed on to
+	 * the next handler.
+	 * @param outputKeys the outputKeys to set
+	 */
+	public void setOutputKeys(String[] outputKeys) {
+		this.outputKeys = outputKeys;
+	}
+
+	/**
+	 * Public setter for the {@link JobRepository} that is needed to manage the
+	 * state of the batch meta domain (jobs, steps, executions) during the life
+	 * of a job.
+	 * 
+	 * @param jobRepository
+	 */
+	@Required
+	public void setJobRepository(JobRepository jobRepository) {
+		this.jobRepository = jobRepository;
+	}
+
+	@Handler
+	public JobExecutionRequest handle(JobExecutionRequest request) {
+
+
+		// Hand off immediately if the job has already failed
+		if (isComplete(request)) {
+			return request;
+		}
+
+		JobExecution jobExecution = request.getJobExecution();
+		JobInstance jobInstance = jobExecution.getJobInstance();
+
+		StepExecution stepExecution = jobExecution.createStepExecution(step);
+		try {
+
+			StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step);
+
+			// Even if it completed successfully we want to pass on the output
+			// attributes, so set up the execution context here if it is
+			// available.
+			if (lastStepExecution != null) {
+				stepExecution.setExecutionContext(lastStepExecution.getExecutionContext());
+			}
+
+			// If it is already complete and not restartable it will simply be
+			// skipped
+			if (shouldStart(lastStepExecution, step)) {
+
+				boolean isRestart = (jobRepository.getStepExecutionCount(jobInstance, step) > 0 && !lastStepExecution
+						.getExitStatus().equals(ExitStatus.FINISHED)) ? true : false;
+
+				if (!isRestart || lastStepExecution == null) {
+					stepExecution.setExecutionContext(getExecutionContextWithInputs(request));
+				}
+
+				step.execute(stepExecution);
+
+			}
+
+			// (the job might actually not be complete, but the stage is).
+			request.setStatus(BatchStatus.COMPLETED);
+
+		}
+		catch (Exception e) {
+			handleFailure(request, e);
+		}
+		catch (Error e) {
+			handleFailure(request, e);
+			throw e;
+		}
+
+		// TODO: could a failure here could cause job to be in inconsistent
+		// state?
+		getMessageWithOutputs(request, stepExecution.getExecutionContext());
+		return request;
+
+	}
+
+	/**
+	 * Clear the message header of the input attributes and add in output
+	 * attributes from the execution context.
+	 * @param request
+	 * @param executionContext
+	 * @return
+	 */
+	private JobExecutionRequest getMessageWithOutputs(JobExecutionRequest request, ExecutionContext executionContext) {
+		for (int i = 0; i < inputKeys.length; i++) {
+			String key = inputKeys[i];
+			request.removeAttribute(key);
+		}
+		for (int i = 0; i < outputKeys.length; i++) {
+			String key = outputKeys[i];
+			request.setAttribute(key, executionContext.get(key));
+		}
+		// TODO: is this safe, or should we build a new message?
+		return request;
+	}
+
+	/**
+	 * Generate a new execution context with all the attributes requested from
+	 * the message (if they exist).
+	 * 
+	 * @return an {@link ExecutionContext}
+	 */
+	private ExecutionContext getExecutionContextWithInputs(JobExecutionRequest request) {
+		ExecutionContext executionContext = new ExecutionContext();
+		for (int i = 0; i < inputKeys.length; i++) {
+			String key = inputKeys[i];
+			Object value = request.getAttribute(key);
+			executionContext.put(key, value);
+		}
+		return executionContext;
+	}
+
+	/**
+	 * @param request
+	 * @return
+	 */
+	private boolean isComplete(JobExecutionRequest request) {
+		return request.getStatus() == BatchStatus.FAILED || request.getStatus() == BatchStatus.STOPPED
+				|| request.getStatus() == BatchStatus.STOPPING;
+	}
+
+	/**
+	 * @param request
+	 * @param e
+	 */
+	private void handleFailure(JobExecutionRequest request, Throwable e) {
+		request.registerThrowable(e);
+		request.setStatus(BatchStatus.FAILED);
+	}
+
+	/*
+	 * TODO: merge this with SimpleJob implementation.
+	 * 
+	 * 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.
+	 */
+	private boolean shouldStart(StepExecution lastStepExecution, Step step) throws JobExecutionException {
+
+		BatchStatus stepStatus;
+		// if the last execution is null, the step has never been executed.
+		if (lastStepExecution == null) {
+			return true;
+		}
+		else {
+			stepStatus = lastStepExecution.getStatus();
+		}
+
+		if (stepStatus == BatchStatus.UNKNOWN) {
+			throw new JobExecutionException("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) {
+			// step is complete, false should be returned, indicating that the
+			// step should not be started
+			return false;
+		}
+
+		if (jobRepository.getStepExecutionCount(lastStepExecution.getJobExecution().getJobInstance(), step) < 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());
+		}
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/job/StepExecutionTimeoutException.java

@@ -0,0 +1,36 @@
+/*
+ * 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.integration.batch.job;
+
+import org.springframework.batch.core.UnexpectedJobExecutionException;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class StepExecutionTimeoutException extends UnexpectedJobExecutionException {
+
+	/**
+	 * Constructs a new instance.
+	 * 
+	 * @param msg the exception message.
+	 * 
+	 */
+	public StepExecutionTimeoutException(String msg) {
+		super(msg);
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/JobLaunchingMessageHandler.java

@@ -0,0 +1,55 @@
+package org.springframework.integration.batch.launch;
+
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobExecutionException;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.integration.annotation.Handler;
+import org.springframework.integration.annotation.MessageEndpoint;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageHandlingException;
+
+/**
+ * Message handler which uses strategies to convert a Message into a job and a
+ * set of job parameters
+ * @author Jonas Partner
+ * @author Dave Syer
+ * 
+ */
+public class JobLaunchingMessageHandler {
+
+	private MessageToJobStrategy messageToJobStrategy;
+
+	private MessageToJobParametersStrategy messageToJobParametersStrategy = new MessagePropertiesToJobParametersStrategy();
+
+	private JobLauncher jobLauncher;
+
+	public JobLaunchingMessageHandler(JobLauncher jobLauncher, MessageToJobStrategy messageToJobStrategy) {
+		super();
+		this.jobLauncher = jobLauncher;
+		this.messageToJobStrategy = messageToJobStrategy;
+	}
+
+	@Handler
+	public JobExecution handle(Message<?> message) {
+		Job job = messageToJobStrategy.getJob(message);
+		JobParameters jobParameters = messageToJobParametersStrategy.getJobParameters(message);
+
+		try {
+			JobExecution execution = jobLauncher.run(job, jobParameters);
+			if (message.getHeader().getReturnAddress() != null) {
+				return execution;
+			}
+			return null;
+		}
+		catch (JobExecutionException e) {
+			throw new MessageHandlingException(message, "Exception executing job ");
+		}
+	}
+
+	public void setMessageToJobParametersStrategy(MessageToJobParametersStrategy messageToJobParametersStrategy) {
+		this.messageToJobParametersStrategy = messageToJobParametersStrategy;
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/JobLaunchingPostReceiveChannelInterceptor.java

@@ -0,0 +1,58 @@
+package org.springframework.integration.batch.launch;
+
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecutionException;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageHandlingException;
+
+/**
+ * Channel interceptor which launches the configured job after the message has
+ * been received
+ * @author Jonas Partner
+ * 
+ */
+public class JobLaunchingPostReceiveChannelInterceptor extends ChannelInterceptorAdapter {
+
+	private JobLauncher jobLauncher;
+
+	private Job job;
+
+	private MessageToJobParametersStrategy messageToJobParametersStrategy = new MessagePropertiesToJobParametersStrategy();
+
+	/**
+	 * 
+	 * @param job The job to launch
+	 * @param jobLauncher
+	 * @param errorHandler
+	 */
+	public JobLaunchingPostReceiveChannelInterceptor(Job job, JobLauncher jobLauncher) {
+		super();
+		this.job = job;
+		this.jobLauncher = jobLauncher;
+	}
+
+	public MessageToJobParametersStrategy getMessageToJobParametersStrategy() {
+		return messageToJobParametersStrategy;
+	}
+
+	public void setMessageToJobParametersStrategy(MessageToJobParametersStrategy messageToJobParametersStrategy) {
+		this.messageToJobParametersStrategy = messageToJobParametersStrategy;
+	}
+
+	@Override
+	public void postReceive(Message<?> message, MessageChannel channel) {
+		JobParameters parameters = messageToJobParametersStrategy.getJobParameters(message);
+		try {
+			jobLauncher.run(job, parameters);
+		}
+		catch (JobExecutionException e) {
+			throw new MessageHandlingException(message, "Excpetion executing job ", e);
+		}
+
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessagePropertiesToJobParametersStrategy.java

@@ -0,0 +1,25 @@
+package org.springframework.integration.batch.launch;
+
+import java.util.Set;
+
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.JobParametersBuilder;
+import org.springframework.integration.message.Message;
+
+/**
+ * Builds an instance of JobParameters from the properties in the message header
+ * @author Jonas Partner
+ *
+ */
+public class MessagePropertiesToJobParametersStrategy implements MessageToJobParametersStrategy {
+
+	public JobParameters getJobParameters(Message<?> message) {
+		JobParametersBuilder parametersBuilder = new JobParametersBuilder();
+		Set<String> propertyNames = message.getHeader().getPropertyNames();
+		for (String key : propertyNames) {
+			parametersBuilder.addString(key, message.getHeader().getProperty(key));
+		}
+		return parametersBuilder.toJobParameters();
+	}
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessageToJobParametersStrategy.java

@@ -0,0 +1,16 @@
+package org.springframework.integration.batch.launch;
+
+import org.springframework.batch.core.JobParameters;
+import org.springframework.integration.message.Message;
+
+
+/**
+ * 
+ * @author Jonas Partner
+ *
+ */
+public interface MessageToJobParametersStrategy {
+	
+	public JobParameters getJobParameters(Message<?> message);
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/MessageToJobStrategy.java

@@ -0,0 +1,16 @@
+package org.springframework.integration.batch.launch;
+
+import org.springframework.batch.core.Job;
+import org.springframework.integration.message.Message;
+
+
+/**
+ * Interface for strategy implementations which convert from a Message to a Spring batch Job
+ * @author Jonas Partner
+ *
+ */
+public interface MessageToJobStrategy{
+	
+	public Job getJob(Message<?> message);
+
+}

spring-batch-integration/src/main/java/org/springframework/integration/batch/launch/StringPayloadAsJobNameStrategy.java

@@ -0,0 +1,32 @@
+package org.springframework.integration.batch.launch;
+
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.configuration.JobLocator;
+import org.springframework.batch.core.repository.NoSuchJobException;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageHandlingException;
+
+/**
+ * Takes the string payload of a message and delegates to a JobLocator
+ * @author Jonas Partner
+ *
+ */
+public class StringPayloadAsJobNameStrategy implements MessageToJobStrategy{
+	
+	private JobLocator jobLocator;
+	
+	public StringPayloadAsJobNameStrategy(JobLocator jobLocator){
+		this.jobLocator = jobLocator;
+	}
+
+	public Job getJob(Message<?> message) {
+		String name = (String)message.getPayload();
+		try {
+			return jobLocator.getJob(name);
+		}
+		catch (NoSuchJobException e) {
+			throw new MessageHandlingException(message, "Could not find job with name " + name, e);
+		}
+	}
+
+}