(Mostly) update for Spring Integration changes

spring-batch-core/src/main/java/org/springframework/batch/core/JobExecution.java

@@ -146,7 +146,7 @@ public class JobExecution extends Entity {
 
 	/**
 	 * Register a step execution with the current job execution.
-	 * @param stepName TODO
+	 * @param stepName the name of the step the new execution is associated with
 	 */
 	public StepExecution createStepExecution(String stepName) {
 		StepExecution stepExecution = new StepExecution(stepName, this);

spring-batch-core/src/main/java/org/springframework/batch/core/step/AbstractStep.java

@@ -353,10 +353,9 @@ public abstract class AbstractStep implements Step, InitializingBean, BeanNameAw
 			exitStatus = new ExitStatus(false, ExitCodeMapper.NO_SUCH_JOB);
 		}
 		else {
-			String message = "";
 			StringWriter writer = new StringWriter();
 			ex.printStackTrace(new PrintWriter(writer));
-			message = writer.toString();
+			String message = writer.toString();
 			exitStatus = ExitStatus.FAILED.addExitDescription(message);
 		}
 

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/AbstractResultQueue.java

@@ -1,111 +0,0 @@
-/*
- * Copyright 2002-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.repeat.support;
-
-import org.springframework.batch.repeat.RepeatException;
-
-/**
- * Abstract implementation that can be extended for both Backport Concurrent and
- * JDK 5 Concurrent.
- * 
- * @author Ben Hale
- * @author Dave Syer
- */
-abstract class AbstractResultQueue extends RepeatInternalStateSupport implements ResultQueue {
-
-	// Arbitrary lock object.
-	private final Object lock = new Object();
-
-	// Counter to monitor the difference between expected and actually collected
-	// results. When this reaches zero there are really no more results.
-	private volatile int count = 0;
-
-	public boolean isExpecting() {
-		synchronized (lock) {
-			// Base the decision about whether we expect more results on a
-			// counter of the number of expected results actually collected.
-			return count > 0;
-		}
-	}
-
-	public void expect() {
-		try {
-			synchronized (lock) {
-				aquireWait();
-				count++;
-			}
-		}
-		catch (InterruptedException e) {
-			Thread.currentThread().interrupt();
-			throw new RepeatException("InterruptedException waiting for to acquire lock on input.");
-		}
-	}
-
-	public void put(ResultHolder holder) {
-		// There should be no need to block here, or to use offer()
-		addResult(holder);
-		// Take from the waits queue now to allow another result to
-		// accumulate. But don't decrement the counter.
-		releaseWait();
-	}
-
-	public ResultHolder take() {
-		ResultHolder value;
-		try {
-			synchronized (lock) {
-				value = takeResult();
-				// Decrement the counter only when the result is collected.
-				count--;
-			}
-		}
-		catch (InterruptedException e) {
-			Thread.currentThread().interrupt();
-			throw new RepeatException("InterruptedException while waiting for result.");
-		}
-		return value;
-	}
-
-	/**
-	 * Acquire permission for one more task on the queue.
-	 * 
-	 * @throws InterruptedException
-	 */
-	protected abstract void aquireWait() throws InterruptedException;
-
-	/**
-	 * Release the permit that we were holding while a task was processed.
-	 */
-	protected abstract void releaseWait();
-
-	/**
-	 * Add a {@link ResultHolder} with a finished result to the queue for
-	 * collection. Should not block. May throw an exception to signal that the
-	 * queue is full, but that would be an unexpected condition.
-	 * 
-	 * @param resultHolder a {@link ResultHolder}
-	 */
-	protected abstract void addResult(ResultHolder resultHolder);
-
-	/**
-	 * Obtain a result from the queue, blocking until one becomes available.
-	 * 
-	 * @return a {@link ResultHolder} with the completed result
-	 * @throws InterruptedException if an interrupt is signalled
-	 */
-	protected abstract ResultHolder takeResult() throws InterruptedException;
-
-}

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/JdkConcurrentResultQueue.java

@@ -20,12 +20,14 @@ import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.Semaphore;
 
+import org.springframework.batch.repeat.RepeatException;
+
 /**
  * An implementation of the {@link ResultQueue} that uses the Java 5 Concurrent Utilities.
  * 
  * @author Ben Hale
  */
-class JdkConcurrentResultQueue extends AbstractResultQueue implements RepeatInternalState {
+class JdkConcurrentResultQueue extends RepeatInternalStateSupport implements ResultQueue {
 
 	// Accumulation of result objects as they finish.
 	private final BlockingQueue<ResultHolder> results;
@@ -33,6 +35,10 @@ class JdkConcurrentResultQueue extends AbstractResultQueue implements RepeatInte
 	// Accumulation of dummy objects flagging expected results in the future.
 	private final Semaphore waits;
 
+	private final Object lock = new Object();
+
+	private volatile int count = 0;
+
 	JdkConcurrentResultQueue(int throttleLimit) {
 		results = new LinkedBlockingQueue<ResultHolder>();
 		waits = new Semaphore(throttleLimit);
@@ -58,4 +64,49 @@ class JdkConcurrentResultQueue extends AbstractResultQueue implements RepeatInte
 		return results.isEmpty();
 	}
 
+	public boolean isExpecting() {
+		synchronized (lock) {
+			// Base the decision about whether we expect more results on a
+			// counter of the number of expected results actually collected.
+			return count > 0;
+		}
+	}
+
+	public void expect() {
+		try {
+			synchronized (lock) {
+				aquireWait();
+				count++;
+			}
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			throw new RepeatException("InterruptedException waiting for to acquire lock on input.");
+		}
+	}
+
+	public void put(ResultHolder holder) {
+		// There should be no need to block here, or to use offer()
+		addResult(holder);
+		// Take from the waits queue now to allow another result to
+		// accumulate. But don't decrement the counter.
+		releaseWait();
+	}
+
+	public ResultHolder take() {
+		ResultHolder value;
+		try {
+			synchronized (lock) {
+				value = takeResult();
+				// Decrement the counter only when the result is collected.
+				count--;
+			}
+		}
+		catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			throw new RepeatException("InterruptedException while waiting for result.");
+		}
+		return value;
+	}
+
 }

spring-batch-infrastructure/src/main/java/org/springframework/batch/repeat/support/ResultQueue.java

@@ -24,6 +24,8 @@ package org.springframework.batch.repeat.support;
  * 
  * @author Dave Syer
  * @author Ben Hale
+ * 
+ * TODO: BlockingQueue with a CompletionService?
  */
 interface ResultQueue extends RepeatInternalState {
 

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

@@ -12,10 +12,10 @@ import org.springframework.batch.item.ItemStream;
 import org.springframework.batch.item.ItemStreamException;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.repeat.ExitStatus;
-import org.springframework.integration.message.BlockingSource;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageTarget;
 import org.springframework.util.Assert;
 
 public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSupport implements ItemWriter<T>, ItemStream {
@@ -28,9 +28,9 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 
 	private static final long DEFAULT_THROTTLE_LIMIT = 6;
 
-	private MessageTarget target;
+	private MessageChannel target;
 
-	private BlockingSource<ChunkResponse> source;
+	private PollableChannel source;
 
 	// TODO: abstract the state or make a factory for this writer?
 	private LocalState localState = new LocalState();
@@ -46,11 +46,13 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 		this.throttleLimit = throttleLimit;
 	}
 
-	public void setSource(BlockingSource<ChunkResponse> source) {
+	// TODO: refactor to ChannelAdapter?
+	public void setInputChannel(PollableChannel source) {
 		this.source = source;
 	}
 
-	public void setTarget(MessageTarget target) {
+	// TODO: re-evaluate the refactor to MessageChannel
+	public void setOutputChannel(MessageChannel target) {
 		this.target = target;
 	}
 
@@ -145,7 +147,8 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 	 * otherwise return null.
 	 */
 	private void getNextResult(long timeout) {
-		Message<ChunkResponse> message = source.receive(timeout);
+		@SuppressWarnings("unchecked")
+		Message<ChunkResponse> message = (Message<ChunkResponse>) source.receive(timeout);
 		if (message != null) {
 			ChunkResponse payload = message.getPayload();
 			Long jobInstanceId = payload.getJobId();
@@ -174,7 +177,6 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
 		}
 
 		public int getSkipCount() {
-			// TODO Auto-generated method stub
 			return stepExecution.getSkipCount();
 		}
 

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

@@ -4,9 +4,11 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.beans.factory.InitializingBean;
-import org.springframework.integration.annotation.Handler;
+import org.springframework.integration.annotation.MessageEndpoint;
+import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.util.Assert;
 
+@MessageEndpoint
 public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, InitializingBean {
 
 	private static final Log logger = LogFactory.getLog(ChunkProcessorChunkHandler.class);
@@ -32,7 +34,7 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, Initializ
 	 * (non-Javadoc)
 	 * @see org.springframework.integration.batch.slave.ChunkHandler#handleChunk(java.util.Collection)
 	 */
-	@Handler
+	@ServiceActivator
 	public ChunkResponse handleChunk(ChunkRequest<S> chunkRequest) {
 
 		logger.debug("Handling chunk: " + chunkRequest);

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

@@ -23,10 +23,10 @@ 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.message.BlockingSource;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageTarget;
 import org.springframework.util.Assert;
 
 /**
@@ -40,9 +40,9 @@ public class MessageOrientedStep extends AbstractStep {
 	 */
 	public static final String WAITING = MessageOrientedStep.class.getName() + ".WAITING";
 
-	private MessageTarget target;
+	private MessageChannel outputChannel;
 
-	private BlockingSource<JobExecutionRequest> source;
+	private PollableChannel source;
 
 	private static int MINUTE = 1000 * 60;
 
@@ -77,11 +77,11 @@ public class MessageOrientedStep extends AbstractStep {
 
 	/**
 	 * Public setter for the target.
-	 * @param target the target to set
+	 * @param outputChannel the target to set
 	 */
 	@Required
-	public void setTarget(MessageTarget target) {
-		this.target = target;
+	public void setOutputChannel(MessageChannel outputChannel) {
+		this.outputChannel = outputChannel;
 	}
 
 	/**
@@ -89,7 +89,7 @@ public class MessageOrientedStep extends AbstractStep {
 	 * @param source the source to set
 	 */
 	@Required
-	public void setSource(BlockingSource<JobExecutionRequest> source) {
+	public void setInputChannel(PollableChannel source) {
 		this.source = source;
 	}
 
@@ -113,7 +113,7 @@ public class MessageOrientedStep extends AbstractStep {
 			executionContext.putString(WAITING, "true");
 			// TODO: need these two lines to be atomic
 			getJobRepository().update(stepExecution);
-			target.send(new GenericMessage<JobExecutionRequest>(request));
+			outputChannel.send(new GenericMessage<JobExecutionRequest>(request));
 			waitForReply(request.getJobId());
 		}
 
@@ -132,7 +132,8 @@ public class MessageOrientedStep extends AbstractStep {
 		while (count++ < maxCount) {
 
 			// TODO: timeout?
-			Message<JobExecutionRequest> message = source.receive(timeout);
+			@SuppressWarnings("unchecked")
+			Message<JobExecutionRequest> message = (Message<JobExecutionRequest>) source.receive(timeout);
 
 			if (message != null) {
 

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

@@ -25,7 +25,7 @@ import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.beans.factory.annotation.Required;
-import org.springframework.integration.annotation.Handler;
+import org.springframework.integration.annotation.ServiceActivator;
 
 /**
  * @author Dave Syer
@@ -58,7 +58,7 @@ public class StepExecutionMessageHandler {
 		this.jobRepository = jobRepository;
 	}
 
-	@Handler
+	@ServiceActivator
 	public JobExecutionRequest handle(JobExecutionRequest request) {
 
 		// Hand off immediately if the job has already failed

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

@@ -6,7 +6,7 @@ import org.springframework.batch.core.JobExecutionException;
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.UnexpectedJobExecutionException;
 import org.springframework.batch.core.launch.JobLauncher;
-import org.springframework.integration.annotation.Handler;
+import org.springframework.integration.annotation.ServiceActivator;
 
 /**
  * Message handler which uses strategies to convert a Message into a job and a
@@ -27,7 +27,7 @@ public class JobLaunchingMessageHandler {
 		this.jobLauncher = jobLauncher;
 	}
 
-	@Handler
+	@ServiceActivator
 	public JobExecution launch(JobLaunchRequest request) {
 		Job job = request.getJob();
 		JobParameters jobParameters = request.getJobParameters();

spring-batch-integration/src/test/java/org/springframework/batch/integration/JobRepositorySupport.java

@@ -35,7 +35,7 @@ public class JobRepositorySupport implements JobRepository {
 	 */
 	public JobExecution createJobExecution(String jobName, JobParameters jobParameters)
 			throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
-		return new JobExecution(new JobInstance(0L, jobParameters, job.getName()));
+		return new JobExecution(new JobInstance(0L, jobParameters, jobName));
 	}
 
 	/* (non-Javadoc)
@@ -76,4 +76,8 @@ public class JobRepositorySupport implements JobRepository {
 	public void update(StepExecution stepExecution) {
 	}
 
+	public boolean isJobInstanceExists(String jobName, JobParameters jobParameters) {
+		return false;
+	}
+
 }

spring-batch-integration/src/test/java/org/springframework/batch/integration/JobSupport.java

@@ -14,7 +14,6 @@ public class JobSupport implements Job {
 	}
 
 	public void execute(JobExecution execution) throws JobExecutionException {
-		// TODO Auto-generated method stub
 	}
 
 	public String getName() {
@@ -22,12 +21,10 @@ public class JobSupport implements Job {
 	}
 
 	public boolean isRestartable() {
-		// TODO Auto-generated method stub
 		return false;
 	}
 	
 	public JobParametersIncrementer getJobParametersIncrementer() {
-		// TODO Auto-generated method stub
 		return null;
 	}
 

spring-batch-integration/src/test/java/org/springframework/batch/integration/SmokeTests.java

@@ -7,10 +7,10 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.integration.annotation.Handler;
 import org.springframework.integration.annotation.MessageEndpoint;
+import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.BlockingSource;
+import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.test.context.ContextConfiguration;
@@ -18,7 +18,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 @ContextConfiguration(locations = "/integration-context.xml")
 @RunWith(SpringJUnit4ClassRunner.class)
-@MessageEndpoint(input = "smokein", output = "smokeout")
+@MessageEndpoint
 public class SmokeTests {
 
 	@Autowired
@@ -27,14 +27,14 @@ public class SmokeTests {
 
 	@Autowired
 	@Qualifier("smokeout")
-	private BlockingSource<String> smokeout;
+	private PollableChannel smokeout;
 
 	// This has to be static because the MessageBus registers the handler
 	// more than once (every time a test instance is created), but only one of
 	// them will get the message.
 	private volatile static int count = 0;
 
-	@Handler
+	@ServiceActivator(inputChannel = "smokein", outputChannel = "smokeout")
 	public String process(String message) {
 		count++;
 		String result = message + ": " + count;
@@ -49,7 +49,8 @@ public class SmokeTests {
 	@Test
 	public void testVanillaSendAndReceive() throws Exception {
 		smokein.send(new GenericMessage<String>("foo"));
-		Message<String> message = smokeout.receive(100);
+		@SuppressWarnings("unchecked")
+		Message<String> message = (Message<String>) smokeout.receive(100);
 		String result = (String) (message == null ? null : message.getPayload());
 		assertEquals("foo: 1", result);
 		assertEquals(1, count);

spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java

@@ -32,6 +32,7 @@ import org.springframework.batch.repeat.ExitStatus;
 import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
@@ -47,7 +48,7 @@ public class ChunkMessageItemWriterIntegrationTests {
 
 	@Autowired
 	@Qualifier("requests")
-	private PollableChannel requests;
+	private MessageChannel requests;
 
 	@Autowired
 	@Qualifier("replies")
@@ -59,7 +60,6 @@ public class ChunkMessageItemWriterIntegrationTests {
 
 	private static long jobCounter;
 
-	@SuppressWarnings("unchecked")
 	@Before
 	public void setUp() {
 
@@ -71,18 +71,13 @@ public class ChunkMessageItemWriterIntegrationTests {
 		factory.setItemWriter(writer);
 		factory.setCommitInterval(4);
 
-		writer.setSource(replies);
-		writer.setTarget(requests);
+		writer.setInputChannel(replies);
+		writer.setOutputChannel(requests);
 
 		TestItemWriter.count = 0;
 		
 		// Drain queues
-		Message<?> message = requests.receive(10);
-		while (message!=null) {
-			System.err.println(message);
-			message = requests.receive(10);
-		}
-		message = replies.receive(10);
+		Message<?> message = replies.receive(10);
 		while (message!=null) {
 			System.err.println(message);
 			message = replies.receive(10);
@@ -92,7 +87,6 @@ public class ChunkMessageItemWriterIntegrationTests {
 
 	@After
 	public void tearDown() {
-		while(requests.receive(10L)!=null) {}
 		while(replies.receive(10L)!=null) {}
 	}
 
@@ -126,7 +120,7 @@ public class ChunkMessageItemWriterIntegrationTests {
 		waitForResults(6, 10);
 
 		assertEquals(6, TestItemWriter.count);
-		assertEquals(6, stepExecution.getItemCount());
+		assertEquals(6, stepExecution.getReadCount());
 
 	}
 
@@ -153,7 +147,7 @@ public class ChunkMessageItemWriterIntegrationTests {
 		waitForResults(8, 10);
 
 		assertEquals(8, TestItemWriter.count);
-		assertEquals(6, stepExecution.getItemCount());
+		assertEquals(6, stepExecution.getReadCount());
 
 	}
 
@@ -185,7 +179,7 @@ public class ChunkMessageItemWriterIntegrationTests {
 		waitForResults(1, 10);
 
 		assertEquals(1, TestItemWriter.count);
-		assertEquals(0, stepExecution.getItemCount());
+		assertEquals(0, stepExecution.getReadCount());
 
 	}
 
@@ -260,7 +254,7 @@ public class ChunkMessageItemWriterIntegrationTests {
 		}
 
 		assertEquals(0, TestItemWriter.count);
-		assertEquals(0, stepExecution.getItemCount());
+		assertEquals(0, stepExecution.getReadCount());
 
 	}
 

spring-batch-integration/src/test/java/org/springframework/batch/integration/file/FileToMessagesJobFactoryBeanTests.java

@@ -20,8 +20,9 @@ import static org.junit.Assert.assertNotNull;
 
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
 
-import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.batch.core.BatchStatus;
@@ -40,8 +41,8 @@ import org.springframework.beans.factory.annotation.Required;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.channel.ThreadLocalChannel;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.util.ReflectionUtils;
 
@@ -53,7 +54,8 @@ public class FileToMessagesJobFactoryBeanTests {
 
 	private static final String FILE_INPUT_PATH = ResourcePayloadAsJobParameterStrategy.FILE_INPUT_PATH;
 	private FileToMessagesJobFactoryBean<FieldSet> factory = new FileToMessagesJobFactoryBean<FieldSet>();
-	private ThreadLocalChannel receiver = new ThreadLocalChannel();
+	private DirectChannel channel = new DirectChannel();
+	private List<FieldSet> receiver = new ArrayList<FieldSet>();
 	private JobRepositorySupport jobRepository;
 
 	@Before
@@ -64,16 +66,15 @@ public class FileToMessagesJobFactoryBeanTests {
 		FlatFileItemReader<FieldSet> itemReader = new FlatFileItemReader<FieldSet>();
 		itemReader.setFieldSetMapper(new PassThroughFieldSetMapper());
 		factory.setItemReader(itemReader);
-		DirectChannel channel = new DirectChannel();
 		factory.setChannel(channel);
-		channel.subscribe(this.receiver);
+		channel.subscribe(new MessageConsumer() {
+			public void onMessage(Message<?> message) {
+				// TODO: Ask Mark: unsafe cast...
+				receiver.add((FieldSet) message.getPayload());
+			}
+		});
 	}
 	
-	@After
-	public void tearDown() {
-		while(receiver.receive(10L)!=null) {}
-	}
-
 	/**
 	 * Test method for
 	 * {@link org.springframework.batch.integration.file.FileToMessagesJobFactoryBean#setBeanName(java.lang.String)}.
@@ -170,7 +171,6 @@ public class FileToMessagesJobFactoryBeanTests {
 		assertEquals(true, factory.isSingleton());
 	}
 	
-	@SuppressWarnings("unchecked")
 	@Test
 	public void testVanillaJobExecution() throws Exception {
 
@@ -183,17 +183,12 @@ public class FileToMessagesJobFactoryBeanTests {
 		assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
 
 		FieldSet payload;
-		Message<FieldSet> message;
 
 		// first line from properties file
-		message = (Message<FieldSet>) receiver.receive(100L);
-		assertNotNull(message);
-		payload = message.getPayload();
+		payload = receiver.get(0);
 		assertNotNull(payload);
 		// second line from properties file
-		message = (Message<FieldSet>) receiver.receive(100L);
-		assertNotNull(message);
-		payload = message.getPayload();
+		payload = receiver.get(1);
 		assertNotNull(payload);
 	}
 

spring-batch-integration/src/test/java/org/springframework/batch/integration/file/ResourceSplitterIntegrationTests.java

@@ -40,7 +40,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  */
 @ContextConfiguration()
 @RunWith(SpringJUnit4ClassRunner.class)
-@MessageEndpoint(input = "resources", output = "requests")
+@MessageEndpoint
 public class ResourceSplitterIntegrationTests {
 
 	@Autowired
@@ -57,7 +57,7 @@ public class ResourceSplitterIntegrationTests {
 	 * The incoming message is a Resource pattern, and it is converted to the
 	 * correct payload type with Spring's default strategy
 	 */
-	@Splitter
+	@Splitter(inputChannel = "resources", outputChannel = "requests")
 	public Resource[] handle(Resource[] message) {
 		List<Resource> list = Arrays.asList(message);
 		System.err.println(list);

spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterTests.java

@@ -29,24 +29,27 @@ import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.ThreadLocalChannel;
-import org.springframework.integration.endpoint.DefaultEndpoint;
+import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageTarget;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.util.ReflectionUtils;
 
 /**
  * @author Dave Syer
- *
+ * 
  */
 public class MessageChannelItemWriterTests {
 
 	/**
-	 * Test method for {@link org.springframework.batch.integration.item.MessageChannelItemWriter#setChannel(org.springframework.integration.channel.MessageChannel)}.
+	 * Test method for
+	 * {@link org.springframework.batch.integration.item.MessageChannelItemWriter#setChannel(org.springframework.integration.channel.MessageChannel)}
+	 * .
 	 */
 	@Test
 	public void testSetChannel() {
-		Method method = ReflectionUtils.findMethod(MessageChannelItemWriter.class, "setChannel", new Class<?>[] {MessageChannel.class});
+		Method method = ReflectionUtils.findMethod(MessageChannelItemWriter.class, "setChannel",
+				new Class<?>[] { MessageChannel.class });
 		assertNotNull(method);
 		Annotation[] annotations = AnnotationUtils.getAnnotations(method);
 		assertEquals(1, annotations.length);
@@ -55,11 +58,9 @@ public class MessageChannelItemWriterTests {
 
 	@Test
 	public void testWrite() throws Exception {
-		DirectChannel channel = new DirectChannel();
 		ThreadLocalChannel receiver = new ThreadLocalChannel();
-		channel.subscribe(receiver);
 		MessageChannelItemWriter<String> writer = new MessageChannelItemWriter<String>();
-		writer.setChannel(channel);
+		writer.setChannel(receiver);
 		writer.write(Collections.singletonList("foo"));
 		Message<?> message = receiver.receive(10);
 		assertNotNull(message);
@@ -69,8 +70,8 @@ public class MessageChannelItemWriterTests {
 	@Test
 	public void testWriteWithRollback() throws Exception {
 		DirectChannel channel = new DirectChannel();
-		channel.subscribe(new MessageTarget() {
-			public boolean send(Message<?> message) {
+		channel.subscribe(new MessageConsumer() {
+			public void onMessage(Message<?> message) {
 				throw new RuntimeException("Planned failure");
 			}
 		});
@@ -88,7 +89,7 @@ public class MessageChannelItemWriterTests {
 	@Test
 	public void testWriteWithRollbackOnEndpoint() throws Exception {
 		DirectChannel channel = new DirectChannel();
-		DefaultEndpoint<MessageHandler> endpoint = new DefaultEndpoint<MessageHandler>(new MessageHandler() {
+		ServiceActivatorEndpoint endpoint = new ServiceActivatorEndpoint(new MessageHandler() {
 			public Message<?> handle(Message<?> message) {
 				throw new RuntimeException("Planned failure");
 			}
@@ -101,7 +102,8 @@ public class MessageChannelItemWriterTests {
 			fail("Expected RuntimeException");
 		}
 		catch (RuntimeException e) {
-			// INT-377: this assertion fails because the exception is wrapped too tightly
+			// INT-377: this assertion fails because the exception is wrapped
+			// too tightly
 			assertEquals("Planned failure", e.getCause().getMessage());
 		}
 	}

spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java

@@ -34,12 +34,12 @@ import org.springframework.batch.integration.JobRepositorySupport;
 import org.springframework.beans.factory.annotation.Required;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.channel.ThreadLocalChannel;
-import org.springframework.integration.message.BlockingSource;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageTarget;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.util.ReflectionUtils;
 
 /**
@@ -56,28 +56,23 @@ public class MessageOrientedStepTests {
 
 	private PollableChannel replyChannel;
 
-	@SuppressWarnings("unchecked")
 	@Before
 	public void createStep() {
 		replyChannel = new ThreadLocalChannel();
 		requestChannel = new DirectChannel();
 		step.setName("step");
-		step.setTarget(requestChannel);
-		step.setSource(replyChannel);
+		step.setOutputChannel(requestChannel);
+		step.setInputChannel(replyChannel);
 		step.setStartLimit(10);
 		step.setJobRepository(new JobRepositorySupport());
 		JobInstance jobInstance = new JobInstance(0L, new JobParameters(), "job");
 		jobExecution = new JobExecution(jobInstance);
 	}
 
-	/**
-	 * Test method for
-	 * {@link org.springframework.batch.integration.job.MessageOrientedStep#setTarget(MessageTarget)}.
-	 */
 	@Test
 	public void testSetRequestChannel() {
-		Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setTarget",
-				new Class<?>[] { MessageTarget.class });
+		Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setOutputChannel",
+				new Class<?>[] { MessageChannel.class });
 		assertNotNull(method);
 		Annotation[] annotations = AnnotationUtils.getAnnotations(method);
 		assertEquals(1, annotations.length);
@@ -86,8 +81,8 @@ public class MessageOrientedStepTests {
 
 	@Test
 	public void testSetReplyChannel() {
-		Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setSource",
-				new Class<?>[] { BlockingSource.class });
+		Method method = ReflectionUtils.findMethod(MessageOrientedStep.class, "setInputChannel",
+				new Class<?>[] { PollableChannel.class });
 		assertNotNull(method);
 		Annotation[] annotations = AnnotationUtils.getAnnotations(method);
 		assertEquals(1, annotations.length);
@@ -96,11 +91,16 @@ public class MessageOrientedStepTests {
 
 	/**
 	 * Test method for
-	 * {@link org.springframework.batch.integration.job.MessageOrientedStep#execute(org.springframework.batch.core.StepExecution)}.
+	 * {@link org.springframework.batch.integration.job.MessageOrientedStep#execute(org.springframework.batch.core.StepExecution)}
+	 * .
 	 * @throws Exception
 	 */
 	@Test
 	public void testExecuteWithTimeout() throws Exception {
+		requestChannel.subscribe(new MessageConsumer() {
+			public void onMessage(Message<?> message) {
+			}
+		});
 		try {
 			step.setExecutionTimeout(1000);
 			step.setPollingInterval(100);
@@ -116,11 +116,11 @@ public class MessageOrientedStepTests {
 
 	@Test
 	public void testVanillaExecute() throws Exception {
-		requestChannel.subscribe(new MessageTarget() {
-			public boolean send(Message<?> message) {
+		requestChannel.subscribe(new MessageConsumer() {
+			public void onMessage(Message<?> message) {
 				JobExecutionRequest jobExecution = (JobExecutionRequest) message.getPayload();
 				jobExecution.setStatus(BatchStatus.COMPLETED);
-				return replyChannel.send(message);
+				replyChannel.send(message);
 			}
 		});
 		step.execute(jobExecution.createStepExecution(step.getName()));
@@ -128,11 +128,11 @@ public class MessageOrientedStepTests {
 
 	@Test
 	public void testExecuteWithFailure() throws Exception {
-		requestChannel.subscribe(new MessageTarget() {
-			public boolean send(Message<?> message) {
+		requestChannel.subscribe(new MessageConsumer() {
+			public void onMessage(Message<?> message) {
 				JobExecutionRequest jobExecution = (JobExecutionRequest) message.getPayload();
 				jobExecution.registerThrowable(new RuntimeException("Planned failure"));
-				return replyChannel.send(message);
+				replyChannel.send(message);
 			}
 		});
 		try {

spring-batch-integration/src/test/java/org/springframework/batch/integration/job/StepExecutionMessageHandlerTests.java

@@ -79,8 +79,8 @@ public class StepExecutionMessageHandlerTests {
 	public void testVanillaHandle() throws Exception {
 		JobRepositorySupport jobRepository = new JobRepositorySupport();
 		StepExecutionMessageHandler handler = createHandler(jobRepository);
-		JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobRepository.createJobExecution(
-				job.getName(), new JobParameters())));
+		JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobRepository.createJobExecution("job",
+				new JobParameters())));
 		assertEquals(1, message.getJobExecution().getStepExecutions().size());
 		assertEquals(BatchStatus.COMPLETED, message.getStatus());
 	}
@@ -89,8 +89,8 @@ public class StepExecutionMessageHandlerTests {
 	public void testHandleWithInputs() throws Exception {
 		JobRepositorySupport jobRepository = new JobRepositorySupport();
 		StepExecutionMessageHandler handler = createHandler(jobRepository);
-		JobExecutionRequest jobExecutionRequest = new JobExecutionRequest(jobRepository.createJobExecution(
-				job.getName(), new JobParameters()));
+		JobExecutionRequest jobExecutionRequest = new JobExecutionRequest(jobRepository.createJobExecution("job",
+				new JobParameters()));
 		jobExecutionRequest.getJobExecution().getExecutionContext().putString("foo", "bar");
 		JobExecutionRequest message = handler.handle(jobExecutionRequest);
 		assertEquals(1, message.getJobExecution().getStepExecutions().size());
@@ -102,8 +102,8 @@ public class StepExecutionMessageHandlerTests {
 	public void testHandleWithInputsAndOutputs() throws Exception {
 		JobRepositorySupport jobRepository = new JobRepositorySupport();
 		StepExecutionMessageHandler handler = createHandler(jobRepository);
-		JobExecutionRequest jobExecutionRequest = new JobExecutionRequest(jobRepository.createJobExecution(
-				job.getName(), new JobParameters()));
+		JobExecutionRequest jobExecutionRequest = new JobExecutionRequest(jobRepository.createJobExecution("job",
+				new JobParameters()));
 		jobExecutionRequest.getJobExecution().getExecutionContext().putString("foo", "bar");
 		// The step has to add the output attribute to the context
 		handler.setStep(new StepSupport("step") {
@@ -122,7 +122,7 @@ public class StepExecutionMessageHandlerTests {
 	public void testHandleFailedJob() throws Exception {
 		JobRepositorySupport jobRepository = new JobRepositorySupport();
 		StepExecutionMessageHandler handler = createHandler(jobRepository);
-		JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters());
+		JobExecution jobExecution = jobRepository.createJobExecution("job", new JobParameters());
 		jobExecution.setStatus(BatchStatus.FAILED);
 		JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution));
 		assertEquals(0, message.getJobExecution().getStepExecutions().size());
@@ -132,8 +132,8 @@ public class StepExecutionMessageHandlerTests {
 	public void testHandleRestart() throws Exception {
 		JobRepositorySupport jobRepository = new JobRepositorySupport() {
 			@Override
-			public StepExecution getLastStepExecution(JobInstance jobInstance, Step step) {
-				StepExecution stepExecution = new StepExecution(step.getName(), new JobExecution(jobInstance));
+			public StepExecution getLastStepExecution(JobInstance jobInstance, String stepName) {
+				StepExecution stepExecution = new StepExecution(stepName, new JobExecution(jobInstance));
 				stepExecution.setStatus(BatchStatus.FAILED);
 				stepExecution.setExecutionContext(new ExecutionContext() {
 					{
@@ -144,20 +144,13 @@ public class StepExecutionMessageHandlerTests {
 				return stepExecution;
 			}
 
-			/*
-			 * (non-Javadoc)
-			 * 
-			 * @seeorg.springframework.integration.batch.JobRepositorySupport#
-			 * getStepExecutionCount(org.springframework.batch.core.JobInstance,
-			 * org.springframework.batch.core.Step)
-			 */
 			@Override
-			public int getStepExecutionCount(JobInstance jobInstance, Step step) {
+			public int getStepExecutionCount(JobInstance jobInstance, String stepName) {
 				return 1;
 			}
 		};
 		StepExecutionMessageHandler handler = createHandler(jobRepository);
-		JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters());
+		JobExecution jobExecution = jobRepository.createJobExecution("job", new JobParameters());
 		JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution));
 		assertNotNull(message);
 		assertEquals(1, jobExecution.getStepExecutions().size());
@@ -169,8 +162,8 @@ public class StepExecutionMessageHandlerTests {
 	public void testHandleRestartAlreadyComplete() throws Exception {
 		JobRepositorySupport jobRepository = new JobRepositorySupport() {
 			@Override
-			public StepExecution getLastStepExecution(JobInstance jobInstance, Step step) {
-				StepExecution stepExecution = new StepExecution(step.getName(), new JobExecution(jobInstance));
+			public StepExecution getLastStepExecution(JobInstance jobInstance, String stepName) {
+				StepExecution stepExecution = new StepExecution(stepName, new JobExecution(jobInstance));
 				stepExecution.setStatus(BatchStatus.COMPLETED);
 				stepExecution.setExecutionContext(new ExecutionContext() {
 					{
@@ -181,7 +174,7 @@ public class StepExecutionMessageHandlerTests {
 			}
 		};
 		StepExecutionMessageHandler handler = createHandler(jobRepository);
-		JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters());
+		JobExecution jobExecution = jobRepository.createJobExecution("job", new JobParameters());
 		JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution));
 		assertNotNull(message);
 		assertEquals(1, jobExecution.getStepExecutions().size());
@@ -196,18 +189,18 @@ public class StepExecutionMessageHandlerTests {
 	public void testHandleRestartStartLimitExceeded() throws Exception {
 		JobRepositorySupport jobRepository = new JobRepositorySupport() {
 			@Override
-			public StepExecution getLastStepExecution(JobInstance jobInstance, Step step) {
-				return new StepExecution(step.getName(), new JobExecution(jobInstance));
+			public StepExecution getLastStepExecution(JobInstance jobInstance, String stepName) {
+				return new StepExecution(stepName, new JobExecution(jobInstance));
 			}
 
 			@Override
-			public int getStepExecutionCount(JobInstance jobInstance, Step step) {
+			public int getStepExecutionCount(JobInstance jobInstance, String stepName) {
 				// sufficiently high restart count
 				return 100;
 			}
 		};
 		StepExecutionMessageHandler handler = createHandler(jobRepository);
-		JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters());
+		JobExecution jobExecution = jobRepository.createJobExecution("job", new JobParameters());
 		JobExecutionRequest message = handler.handle(new JobExecutionRequest(jobExecution));
 		assertNotNull(message);
 		assertEquals(1, jobExecution.getStepExecutions().size());

spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests.java

@@ -16,12 +16,12 @@ import org.springframework.batch.core.JobParametersBuilder;
 import org.springframework.batch.integration.JobSupport;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageDeliveryException;
 import org.springframework.integration.message.MessageHeaders;
-import org.springframework.integration.message.MessageTarget;
+import org.springframework.integration.message.MessagingException;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -32,7 +32,7 @@ public class JobLaunchingMessageHandlerIntegrationTests {
 
 	@Autowired
 	@Qualifier("requests")
-	private MessageTarget requestChannel;
+	private MessageChannel requestChannel;
 
 	@Autowired
 	@Qualifier("response")
@@ -54,9 +54,9 @@ public class JobLaunchingMessageHandlerIntegrationTests {
 		try {
 			requestChannel.send(trigger);
 		}
-		catch (MessageDeliveryException e) {
+		catch (MessagingException e) {
 			String message = e.getMessage();
-			assertTrue("Wrong message: " + message, message.contains("reply target"));
+			assertTrue("Wrong message: " + message, message.contains("reply channel"));
 		}
 		Message<JobExecution> executionMessage = (Message<JobExecution>) responseChannel.receive(1000);
 
@@ -79,7 +79,7 @@ public class JobLaunchingMessageHandlerIntegrationTests {
 
 		assertNotNull("No response received", executionMessage);
 		JobExecution execution = executionMessage.getPayload();
-		assertNotNull("JobExectuion not returned", execution);
+		assertNotNull("JobExecution not returned", execution);
 	}
 
 }

spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobRequestConverter.java

@@ -19,7 +19,7 @@ import java.util.Properties;
 
 import org.springframework.batch.core.converter.DefaultJobParametersConverter;
 import org.springframework.batch.integration.JobSupport;
-import org.springframework.integration.annotation.Handler;
+import org.springframework.integration.annotation.ServiceActivator;
 
 /**
  * @author Dave Syer
@@ -27,7 +27,7 @@ import org.springframework.integration.annotation.Handler;
  */
 public class JobRequestConverter {
 
-	@Handler
+	@ServiceActivator
 	public JobLaunchRequest convert(String jobName) {
 		// TODO: get these from message header
 		Properties properties = new Properties();

spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java

@@ -17,6 +17,9 @@ import org.springframework.aop.support.DefaultPointcutAdvisor;
 import org.springframework.aop.support.NameMatchMethodPointcut;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.support.ListItemReader;
+import org.springframework.batch.repeat.interceptor.RepeatOperationsInterceptor;
+import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
+import org.springframework.batch.repeat.support.RepeatTemplate;
 import org.springframework.batch.retry.interceptor.MethodArgumentsKeyGenerator;
 import org.springframework.batch.retry.interceptor.MethodInvocationRecoverer;
 import org.springframework.batch.retry.interceptor.StatefulRetryOperationsInterceptor;
@@ -24,11 +27,11 @@ import org.springframework.batch.support.transaction.ResourcelessTransactionMana
 import org.springframework.batch.support.transaction.TransactionAwareProxyFactory;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.DirectChannel;
-import org.springframework.integration.dispatcher.PollingDispatcher;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.endpoint.SourcePoller;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageExchangeTemplate;
-import org.springframework.integration.message.MessageTarget;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.message.PollableSource;
 import org.springframework.integration.scheduling.PollingSchedule;
 import org.springframework.integration.scheduling.SchedulableTask;
@@ -36,6 +39,8 @@ import org.springframework.integration.scheduling.TaskScheduler;
 import org.springframework.integration.scheduling.spi.ProviderTaskScheduler;
 import org.springframework.integration.scheduling.spi.SimpleScheduleServiceProvider;
 import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
+import org.springframework.transaction.interceptor.TransactionInterceptor;
 import org.springframework.util.StringUtils;
 
 public class PollableSourceRetryTests {
@@ -70,16 +75,16 @@ public class PollableSourceRetryTests {
 		list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k")));
 		int beforeCount = list.size();
 
-		MessageTarget handler = new MessageTarget() {
-			public boolean send(Message<?> message) {
+		MessageConsumer handler = new MessageConsumer() {
+			public void onMessage(Message<?> message) {
 				Object payload = message.getPayload();
 				logger.debug("Handling: " + payload);
-				return processed.add((String) payload);
+				processed.add((String) payload);
 			}
 		};
 		PollableSource<Object> source = getPollableSource(list);
-		MessageTarget target = getChannel(handler);
-		PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 1);
+		MessageChannel target = getChannel(handler);
+		SourcePoller trigger = getSourcePoller(source, target, transactionManager, 1);
 		TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
 
 		waitForResults(scheduler, 2, 40);
@@ -98,8 +103,8 @@ public class PollableSourceRetryTests {
 		list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k")));
 		int beforeCount = list.size();
 
-		MessageTarget handler = new MessageTarget() {
-			public boolean send(Message<?> message) {
+		MessageConsumer handler = new MessageConsumer() {
+			public void onMessage(Message<?> message) {
 				Object payload = message.getPayload();
 				logger.debug("Handling: " + payload);
 				processed.add((String) payload);
@@ -107,8 +112,8 @@ public class PollableSourceRetryTests {
 			}
 		};
 		PollableSource<Object> source = getPollableSource(list);
-		MessageTarget target = getChannel(handler);
-		PollingDispatcher trigger = getPollingDispatcher(source, target, null, 1);
+		MessageChannel target = getChannel(handler);
+		SourcePoller trigger = getSourcePoller(source, target, null, 1);
 		TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
 
 		waitForResults(scheduler, 2, 20);
@@ -129,8 +134,8 @@ public class PollableSourceRetryTests {
 		list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k")));
 		int beforeCount = list.size();
 
-		MessageTarget handler = new MessageTarget() {
-			public boolean send(Message<?> message) {
+		MessageConsumer handler = new MessageConsumer() {
+			public void onMessage(Message<?> message) {
 				Object payload = message.getPayload();
 				logger.debug("Handling: " + payload);
 				processed.add((String) payload);
@@ -138,8 +143,8 @@ public class PollableSourceRetryTests {
 			}
 		};
 		PollableSource<Object> source = getPollableSource(list);
-		MessageTarget target = getChannel(handler);
-		PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 1);
+		MessageChannel target = getChannel(handler);
+		SourcePoller trigger = getSourcePoller(source, target, transactionManager, 1);
 		TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
 
 		waitForResults(scheduler, 2, 40);
@@ -162,21 +167,20 @@ public class PollableSourceRetryTests {
 		list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k")));
 		int beforeCount = list.size();
 
-		MessageTarget handler = new MessageTarget() {
-			public boolean send(Message<?> message) {
+		MessageConsumer handler = new MessageConsumer() {
+			public void onMessage(Message<?> message) {
 				Object payload = message.getPayload();
 				logger.debug("Handling: " + payload);
-				boolean result = processed.add((String) payload);
+				processed.add((String) payload);
 				if ("fail".equals(payload)) {
 					throw new RuntimeException("Planned failure: " + payload);
 				}
-				return result;
 			}
 		};
 
 		PollableSource<Object> source = getPollableSource(list);
-		MessageTarget target = getChannel(handler);
-		PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 1);
+		MessageChannel target = getChannel(handler);
+		SourcePoller trigger = getSourcePoller(source, target, transactionManager, 1);
 		TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
 
 		waitForResults(scheduler, 5, 50);
@@ -202,22 +206,24 @@ public class PollableSourceRetryTests {
 		list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k")));
 		int beforeCount = list.size();
 
-		MessageTarget handler = new MessageTarget() {
-			public boolean send(Message<?> message) {
+		MessageConsumer handler = new MessageConsumer() {
+			public void onMessage(Message<?> message) {
 				Object payload = message.getPayload();
 				logger.debug("Handling: " + payload);
-				boolean result = processed.add((String) payload);
+				processed.add((String) payload);
 				if ("fail".equals(payload)) {
 					throw new RuntimeException("Planned failure: " + payload);
 				}
-				return result;
 			}
 		};
 
 		PollableSource<Object> source = getPollableSource(list);
-		MessageTarget target = getChannel(handler);
-		PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 3);
-		TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
+		MessageChannel target = getChannel(handler);
+		SourcePoller trigger = getSourcePoller(source, target, null, 1);
+		SchedulableTask task = (SchedulableTask) getProxy(trigger, SchedulableTask.class, new Advice[] {
+				new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource()),
+				getRepeatOperationsInterceptor(3) }, "run");
+		TaskScheduler scheduler = getSchedulerWithErrorHandler(task);
 
 		waitForResults(scheduler, 6, 100);
 
@@ -242,29 +248,29 @@ public class PollableSourceRetryTests {
 		list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k")));
 		int beforeCount = list.size();
 
-		MessageTarget handler = new MessageTarget() {
-			public boolean send(Message<?> message) {
+		MessageConsumer handler = new MessageConsumer() {
+			public void onMessage(Message<?> message) {
 				if (message == null) {
-					return false;
+					return;
 				}
 				Object payload = message.getPayload();
 				logger.debug("Handling: " + payload);
-				boolean result = processed.add((String) payload);
+				processed.add((String) payload);
 				// INT-184 this won't work if it is a "real" handler that throws
 				// MessageHandlingException
 				if ("fail".equals(payload)) {
 					throw new RuntimeException("Planned failure: " + payload);
 				}
-				return result;
 			}
 		};
 
 		PollableSource<Object> source = getPollableSource(list);
-		MessageTarget target = getChannel(handler);
+		MessageChannel target = getChannel(handler);
 		// this was the old dispatch advice chain
-		target = (MessageTarget) getProxy(target, MessageTarget.class,
+		target = (MessageChannel) getProxy(target, MessageChannel.class,
 				new Advice[] { getRetryOperationsInterceptor(methodArgumentsKeyGenerator) }, "send");
-		PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 1);
+		SourcePoller trigger = getSourcePoller(source, target, transactionManager, 1);
+
 		TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
 
 		waitForResults(scheduler, 4, 40);
@@ -290,34 +296,37 @@ public class PollableSourceRetryTests {
 		list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,fail,c,d,e,f,g,h,j,k")));
 		int beforeCount = list.size();
 
-		MessageTarget handler = new MessageTarget() {
-			public boolean send(Message<?> message) {
+		MessageConsumer handler = new MessageConsumer() {
+			public void onMessage(Message<?> message) {
 				Object payload = message.getPayload();
 				logger.debug("Handling: " + payload);
-				boolean result = processed.add((String) payload);
+				processed.add((String) payload);
 				if ("fail".equals(payload)) {
 					throw new RuntimeException("Planned failure: " + payload);
 				}
-				return result;
 			}
 		};
 
 		PollableSource<Object> source = getPollableSource(list);
-		MessageTarget target = getChannel(handler);
+		MessageChannel target = getChannel(handler);
+
 		// this was the old dispatch advice chain
-		target = (MessageTarget) getProxy(target, MessageTarget.class,
+		target = (MessageChannel) getProxy(target, MessageChannel.class,
 				new Advice[] { getRetryOperationsInterceptor(methodArgumentsKeyGenerator) }, "send");
-		PollingDispatcher trigger = getPollingDispatcher(source, target, transactionManager, 3);
-		TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
+		SourcePoller trigger = getSourcePoller(source, target, null, 1);
+		SchedulableTask task = (SchedulableTask) getProxy(trigger, SchedulableTask.class, new Advice[] {
+			new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource()),
+			getRepeatOperationsInterceptor(3) }, "run");
+		TaskScheduler scheduler = getSchedulerWithErrorHandler(task);
 
 		waitForResults(scheduler, 6, 100);
 		System.err.println(processed);
 		System.err.println(list);
 
-		assertEquals(6, processed.size());
 		assertFalse("No messages got to processor", processed.isEmpty());
-		// One roll back and then start again with a,b,d,e
-		assertEquals(beforeCount - 5, list.size());
+		assertEquals(7, processed.size());
+		// 6 items were removed from the list
+		assertEquals(beforeCount - 6, list.size());
 		assertEquals("a", processed.get(0));
 		assertEquals("fail", processed.get(1));
 		// retry makes it fail once then recover...
@@ -327,26 +336,18 @@ public class PollableSourceRetryTests {
 
 	}
 
-	private PollingDispatcher getPollingDispatcher(PollableSource<Object> source, MessageTarget target,
-			PlatformTransactionManager transactionManager, int commitInterval) {
-		MessageExchangeTemplate template = getExchangeTemplate(transactionManager);
-		PollingDispatcher dispatcher = new PollingDispatcher(source, new PollingSchedule(100), null, template);
-		dispatcher.setMaxMessagesPerPoll(commitInterval);
-		dispatcher.subscribe(target);
-		return dispatcher;
+	private SourcePoller getSourcePoller(PollableSource<Object> source, MessageChannel channel,
+			PlatformTransactionManager transactionManager, int maxMessagesPerPoll) {
+		SourcePoller poller = new SourcePoller(source, channel, new PollingSchedule(100));
+		poller.setTransactionManager(transactionManager);
+		poller.setMaxMessagesPerPoll(maxMessagesPerPoll);
+		return poller;
 	}
 
-	private MessageExchangeTemplate getExchangeTemplate(PlatformTransactionManager transactionManager) {
-		MessageExchangeTemplate template = new MessageExchangeTemplate();
-		template.setTransactionManager(transactionManager);
-		template.afterPropertiesSet();
-		return template;
-	}
-
-	private DirectChannel getChannel(MessageTarget target) {
+	private DirectChannel getChannel(MessageConsumer handler) {
 		DirectChannel channel = new DirectChannel();
 		channel.setBeanName("input");
-		channel.subscribe(target);
+		channel.subscribe(handler);
 		return channel;
 	}
 
@@ -398,7 +399,8 @@ public class PollableSourceRetryTests {
 	 * @param methodArgumentsKeyGenerator
 	 * @return
 	 */
-	private StatefulRetryOperationsInterceptor getRetryOperationsInterceptor(MethodArgumentsKeyGenerator methodArgumentsKeyGenerator) {
+	private StatefulRetryOperationsInterceptor getRetryOperationsInterceptor(
+			MethodArgumentsKeyGenerator methodArgumentsKeyGenerator) {
 		StatefulRetryOperationsInterceptor advice = new StatefulRetryOperationsInterceptor();
 		advice.setRecoverer(new MethodInvocationRecoverer<Boolean>() {
 			@SuppressWarnings("unchecked")
@@ -416,21 +418,18 @@ public class PollableSourceRetryTests {
 		return advice;
 	}
 
+	private RepeatOperationsInterceptor getRepeatOperationsInterceptor(int commitInterval) {
+		RepeatOperationsInterceptor advice = new RepeatOperationsInterceptor();
+		RepeatTemplate repeatTemplate = new RepeatTemplate();
+		repeatTemplate.setCompletionPolicy(new SimpleCompletionPolicy(commitInterval));
+		advice.setRepeatOperations(repeatTemplate);
+		return advice;
+	}
 
 	/**
 	 * @param commitInterval
 	 * @return
 	 */
-	// private RepeatOperationsInterceptor getRepeatOperationsInterceptor(int
-	// commitInterval) {
-	// RepeatOperationsInterceptor advice = new RepeatOperationsInterceptor();
-	// RepeatTemplate repeatTemplate = new RepeatTemplate();
-	// repeatTemplate.setCompletionPolicy(new
-	// SimpleCompletionPolicy(commitInterval));
-	// advice.setRepeatOperations(repeatTemplate);
-	// return advice;
-	// }
-
 	private Object getProxy(Object target, Class<?> intf, Advice[] advices, String methodName) {
 		ProxyFactory factory = new ProxyFactory(target);
 		for (int i = 0; i < advices.length; i++) {

spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<beans:beans xmlns="http://www.springframework.org/schema/integration"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
-	xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
+<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:beans="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:tx="http://www.springframework.org/schema/tx"
 	xsi:schemaLocation="http://www.springframework.org/schema/beans
                                  http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
                                  http://www.springframework.org/schema/integration
@@ -10,18 +10,14 @@
                                  http://www.springframework.org/schema/context/spring-context-2.5.xsd
                                  http://www.springframework.org/schema/tx
                                  http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
-
-	<message-bus/>
-
+	<message-bus />
 	<annotation-driven />
-
 	<channel id="requests" />
-	<channel id="replies" />
-
-	<beans:bean id="transactionManager"
-		class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" />
+	<channel id="replies">
+		<queue capacity="UNBOUNDED" />
+	</channel>
+	<beans:bean id="transactionManager" class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" />
 	<tx:annotation-driven />
-
 	<beans:bean id="chunkHandler" class="org.springframework.batch.integration.chunk.ChunkProcessorChunkHandler">
 		<beans:property name="chunkProcessor">
 			<beans:bean class="org.springframework.batch.integration.chunk.SimpleChunkProcessor">
@@ -34,7 +30,5 @@
 			</beans:bean>
 		</beans:property>
 	</beans:bean>
-
-	<service-activator input-channel="requests" output-channel="replies" ref="chunkHandler" method="handleChunk" />
-
+	<service-activator input-channel="requests" output-channel="replies" ref="chunkHandler" />
 </beans:beans>

spring-batch-integration/src/test/resources/org/springframework/batch/integration/item/MessageChannelItemWriterIntegrationTests-context.xml

@@ -12,7 +12,9 @@
                                  http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
 		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
 	<integration:message-bus/>
-	<integration:channel id="requests" />
+	<integration:channel id="requests">
+		<integration:queue capacity="UNBOUNDED"/>
+	</integration:channel>
 	<bean id="itemWriter" class="org.springframework.batch.integration.item.MessageChannelItemWriter">
 		<property name="channel" ref="requests"/>
 	</bean>

spring-batch-integration/src/test/resources/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests-context.xml

@@ -1,11 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<beans xmlns="http://www.springframework.org/schema/beans"
-	xmlns:integration="http://www.springframework.org/schema/integration"
-	xmlns:context="http://www.springframework.org/schema/context"
-	xmlns:aop="http://www.springframework.org/schema/aop"
-	xmlns:tx="http://www.springframework.org/schema/tx"
-	xmlns:p="http://www.springframework.org/schema/p"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+<beans xmlns="http://www.springframework.org/schema/beans" xmlns:integration="http://www.springframework.org/schema/integration"
+	xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"
+	xmlns:tx="http://www.springframework.org/schema/tx" xmlns:p="http://www.springframework.org/schema/p" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	xsi:schemaLocation="
 		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
 		http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
@@ -14,35 +10,31 @@
                                  http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
 		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
 	<import resource="classpath:/simple-job-launcher-context.xml" />
-	<integration:annotation-driven/>
-	<integration:message-bus/>
+	<integration:annotation-driven />
+	<integration:message-bus />
 	<integration:channel id="requests" />
 	<integration:channel id="replies">
-		<integration:queue capacity="UNBOUNDED"/>
+		<integration:queue capacity="UNBOUNDED" />
 	</integration:channel>
 	<bean id="job" parent="simpleJob">
 		<property name="steps">
-			<bean id="stepController"
-				class="org.springframework.batch.integration.job.MessageOrientedStep">
-				<property name="target" ref="requests" />
-				<property name="source" ref="replies" />
+			<bean id="stepController" class="org.springframework.batch.integration.job.MessageOrientedStep">
+				<property name="outputChannel" ref="requests" />
+				<property name="inputChannel" ref="replies" />
 				<property name="jobRepository" ref="jobRepository" />
 			</bean>
 		</property>
 		<property name="jobRepository" ref="jobRepository" />
 	</bean>
-	<bean id="step"
-		class="org.springframework.batch.integration.job.StepExecutionMessageHandler">
+	<bean id="step" class="org.springframework.batch.integration.job.StepExecutionMessageHandler">
 		<property name="step">
 			<bean parent="taskletStep">
 				<property name="tasklet">
-					<bean
-						class="org.springframework.batch.integration.job.TestTasklet" />
+					<bean class="org.springframework.batch.integration.job.TestTasklet" />
 				</property>
 			</bean>
 		</property>
 		<property name="jobRepository" ref="jobRepository" />
 	</bean>
-	<integration:service-activator ref="step" method="handle" input-channel="requests"
-		 output-channel="replies" />
+	<integration:service-activator ref="step" input-channel="requests" output-channel="replies" />
 </beans>

spring-batch-integration/src/test/resources/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests-context.xml

@@ -1,9 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<beans xmlns="http://www.springframework.org/schema/beans"
-	xmlns:integration="http://www.springframework.org/schema/integration"
+<beans xmlns="http://www.springframework.org/schema/beans" xmlns:integration="http://www.springframework.org/schema/integration"
 	xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"
-	xmlns:tx="http://www.springframework.org/schema/tx" xmlns:p="http://www.springframework.org/schema/p"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:tx="http://www.springframework.org/schema/tx" xmlns:p="http://www.springframework.org/schema/p" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	xsi:schemaLocation="
 		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
 		http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
@@ -11,34 +9,25 @@
 		http://www.springframework.org/schema/integration
                                  http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
 		http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
-
 	<import resource="classpath:simple-job-launcher-context.xml" />
-
-	<integration:message-bus/>
+	<integration:message-bus />
 	<integration:annotation-driven />
 	<integration:channel id="requests" />
-	<integration:channel id="jobs" />
 	<integration:channel id="response">
-		<integration:queue capacity="UNBOUNDED"/>
+		<integration:queue capacity="UNBOUNDED" />
 	</integration:channel>
 
-	<integration:service-activator input-channel="requests" ref="jobLaunchingHandler"/>
-	<integration:service-activator input-channel="none" output-channel="jobs" ref="jobRequestConverter" />
-
-	<bean id="jobRequestConverter" class="org.springframework.batch.integration.launch.JobRequestConverter" />
+	<integration:service-activator input-channel="requests" ref="jobLaunchingHandler" method="launch" />
 
 	<bean id="jobLaunchingHandler" class="org.springframework.batch.integration.launch.JobLaunchingMessageHandler">
 		<constructor-arg ref="jobLauncher" />
 	</bean>
-
 	<bean id="testJob" parent="simpleJob">
 		<property name="steps" ref="step" />
 	</bean>
-
 	<bean id="step" parent="taskletStep">
 		<property name="tasklet">
 			<bean class="org.springframework.batch.integration.job.TestTasklet" />
 		</property>
 	</bean>
-
 </beans>