diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptor.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptor.java
new file mode 100644
index 000000000..b4f6eeff7
--- /dev/null
+++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptor.java
@@ -0,0 +1,76 @@
+package org.springframework.batch.integration.chunk;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.channel.ChannelInterceptor;
+import org.springframework.integration.channel.ThreadLocalChannel;
+import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessageChannel;
+import org.springframework.integration.message.MessageSource;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link ChannelInterceptor} that turns a pollable channel into a "pass-thru channel": if a client calls
+ * receive() on the channel it will delegate to a {@link MessageSource} to pull the message directly from
+ * an external source. This is particularly useful in combination with a {@link ThreadLocalChannel}, in which case the
+ * receive() can join a transaction which was started by the caller.
+ *
+ * @author Dave Syer
+ *
+ */
+public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter implements InitializingBean {
+
+ private static Log logger = LogFactory.getLog(MessageSourcePollerInterceptor.class);
+
+ private MessageSource source;
+
+ /**
+ * Convenient default constructor for configuration purposes.
+ */
+ public MessageSourcePollerInterceptor() {
+ }
+
+ /**
+ * @param source a message source to poll for messages on receive.
+ */
+ public MessageSourcePollerInterceptor(MessageSource source) {
+ this.source = source;
+ }
+
+ /**
+ * Asserts that mandatory properties are set.
+ * @see InitializingBean#afterPropertiesSet()
+ */
+ public void afterPropertiesSet() throws Exception {
+ Assert.state(source != null, "A MessageSource must be provided");
+ }
+
+ /**
+ * @param source a message source to poll for messages on receive.
+ */
+ public void setMessageSource(MessageSource source) {
+ this.source = source;
+ }
+
+ /**
+ * Receive from the {@link MessageSource} and send immediately to the input channel, so that the call that we are
+ * intercepting always a message to receive.
+ *
+ * @see ChannelInterceptorAdapter#preReceive(MessageChannel)
+ */
+ @Override
+ public boolean preReceive(MessageChannel channel) {
+ Message message = source.receive();
+ if (message != null) {
+ channel.send(message);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Sent " + message + " to channel " + channel.getName());
+ }
+ return true;
+ }
+ return true;
+ }
+
+}
diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptorTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptorTests.java
new file mode 100644
index 000000000..5d043f67c
--- /dev/null
+++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptorTests.java
@@ -0,0 +1,55 @@
+package org.springframework.batch.integration.chunk;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+import org.springframework.integration.channel.ThreadLocalChannel;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.message.MessageSource;
+
+public class MessageSourcePollerInterceptorTests {
+
+ @Test(expected = IllegalStateException.class)
+ public void testMandatoryPropertiesUnset() throws Exception {
+ MessageSourcePollerInterceptor interceptor = new MessageSourcePollerInterceptor();
+ interceptor.afterPropertiesSet();
+ }
+
+ @Test
+ public void testMandatoryPropertiesSetViaConstructor() throws Exception {
+ MessageSourcePollerInterceptor interceptor = new MessageSourcePollerInterceptor(new TestMessageSource("foo"));
+ interceptor.afterPropertiesSet();
+ }
+
+ @Test
+ public void testMandatoryPropertiesSet() throws Exception {
+ MessageSourcePollerInterceptor interceptor = new MessageSourcePollerInterceptor();
+ interceptor.setMessageSource(new TestMessageSource("foo"));
+ interceptor.afterPropertiesSet();
+ }
+
+ @Test
+ public void testPreReceive() throws Exception {
+ MessageSourcePollerInterceptor interceptor = new MessageSourcePollerInterceptor(new TestMessageSource("foo"));
+ ThreadLocalChannel channel = new ThreadLocalChannel();
+ assertTrue(interceptor.preReceive(channel));
+ assertEquals("foo", channel.receive(10L).getPayload());
+ }
+
+ private static class TestMessageSource implements MessageSource {
+
+ private String payload;
+
+ public TestMessageSource(String payload) {
+ super();
+ this.payload = payload;
+ }
+
+ public Message receive() {
+ return new GenericMessage(payload);
+ }
+ }
+
+}
diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml
index bcefb0278..a5df7b030 100644
--- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml
+++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml
@@ -2,8 +2,9 @@
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-
+