BATCHADM-49: Use a ChannelInterceptor to turn a ThreadLocalChannel into a pass-thru channel. Keeps JMS receive in the caller's transaction.

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
+ * <code>receive()</code> 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
+ * <code>receive()</code> 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;
+	}
+
+}

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<String> {
+
+		private String payload;
+
+		public TestMessageSource(String payload) {
+			super();
+			this.payload = payload;
+		}
+
+		public Message<String> receive() {
+			return new GenericMessage<String>(payload);
+		}
+	}
+
+}

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

@@ -2,8 +2,9 @@
 <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	xmlns:integration="http://www.springframework.org/schema/integration" xmlns:batch="http://www.springframework.org/schema/batch"
 	xmlns:context="http://www.springframework.org/schema/context" xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
-	xmlns:jms="http://www.springframework.org/schema/jms"
-	xsi:schemaLocation="http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd
+	xmlns:jms="http://www.springframework.org/schema/jms" xmlns:aop="http://www.springframework.org/schema/aop"
+	xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
+		http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd
 		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
 		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-1.0.xsd
 		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
@@ -54,14 +55,27 @@
 
 	<int-jms:outbound-channel-adapter connection-factory="connectionFactory" channel="requests"
 		destination-name="requests" />
-	<int-jms:message-driven-channel-adapter connection-factory="connectionFactory"
-		channel="replies" destination-name="replies" transaction-manager="transactionManager" acknowledge="transacted"/>
 	<integration:channel id="requests" />
-	<integration:channel id="replies">
-		<integration:queue />
-	</integration:channel>
+	<integration:thread-local-channel id="replies">
+		<integration:interceptors>
+			<bean id="pollerInterceptor" class="org.springframework.batch.integration.chunk.MessageSourcePollerInterceptor">
+				<property name="messageSource">
+					<bean class="org.springframework.integration.jms.JmsDestinationPollingSource">
+						<constructor-arg>
+							<bean class="org.springframework.jms.core.JmsTemplate">
+								<property name="connectionFactory" ref="connectionFactory" />
+								<property name="defaultDestinationName" value="replies" />
+								<property name="receiveTimeout" value="100" />
+							</bean>
+						</constructor-arg>
+					</bean>
+				</property>
+			</bean>
+		</integration:interceptors>
+	</integration:thread-local-channel>
 
-	<jms:listener-container connection-factory="connectionFactory" transaction-manager="transactionManager" acknowledge="transacted">
+	<jms:listener-container connection-factory="connectionFactory" transaction-manager="transactionManager"
+		acknowledge="transacted">
 		<jms:listener destination="requests" response-destination="replies" ref="chunkHandler" method="handleChunk" />
 	</jms:listener-container>