INT-2683, INT-2684 JmsOutboundGateway improvements

Added support for caching reply consumers when reply queue is Temporary Queue
Added support for reuse of temporary Reply queues in the JmsOutboundGateway

Added check in ChannelPublishingJmsMessageListener to catch IllegalAccessExceptions
when Session is closed

INT-2683 first round of PR coomments

INT-2683 polish

INT-2683 polishing
code restructuring, other optimizations

INT-2683 modified 'messageCorrelationId' generation logic to be the object identity of the jmsReques as its 3-4 times fatster then String.valueOf(System.currentTimeMillis() +
-                String.valueOf(System.nanoTime())

INT-2683
polished JOG to ensure that it properly manages the correlation id when used in the pipeline
added various combinations of pipeline tests

INT-2683 added copyrights to tests

INT-2683 polishing

INT-2683 polishing

INT-2683 added more pipeline tests

INT-2683 added more tests

INT-2683 polishing test

INT-2683 Polishing

INT-2683 More Pipeline Tests

INT-2683 Don't Propagate CorrelationID

Message correlation is internal to the JOG. Cannot
propagate an existing correlation id because some
downstream JOG may end up correlating (selecting) on
the same key.

INT-2683 Polishing

Fix whitespace - replace spaces with tabs.

INT-2683 polishing
reverted 2.1 schema back to its master state, fixed the typo in the test

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsWithMarshallingMessageConverterTests.java

@@ -44,7 +44,7 @@ import org.springframework.oxm.XmlMappingException;
  * @author Oleg Zhurakousky
  */
 public class JmsWithMarshallingMessageConverterTests {
-	
+
 	@Test
 	@SuppressWarnings("unchecked")
 	public void demoWithMarshallingConverter() {
@@ -58,7 +58,6 @@ public class JmsWithMarshallingMessageConverterTests {
 		MessageHeaders headers = replyMessage.getHeaders();
 		// check for couple of JMS headers, make sure they are present
 		assertNotNull(headers.get("jms_redelivered"));
-		assertNotNull(headers.get("jms_correlationId"));
 		assertEquals("HELLO", replyMessage.getPayload());
 	}
 
@@ -86,7 +85,7 @@ public class JmsWithMarshallingMessageConverterTests {
 
 		public boolean supports(Class<?> clazz) {
 			return true;
-		}	
+		}
 	}
 
 

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2002-2012 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.jms.request_reply;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.gateway.RequestReplyExchanger;
+import org.springframework.integration.jms.config.ActiveMqTestUtils;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.util.StopWatch;
+/**
+ * @author Oleg Zhurakousky
+ */
+public class MiscellaneousTests {
+
+	/**
+	 * Asserts that receive-timeout is honored even if
+	 * requests (once in process), takes less then receive-timeout value
+	 * when requests are queued up (e.g., single consumer receiver)
+	 */
+	@Test
+	public void testTimeoutHonoringWhenRequestsQueuedUp() throws Exception{
+		ActiveMqTestUtils.prepare();
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("honor-timeout.xml", this.getClass());
+		final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
+		final CountDownLatch latch = new CountDownLatch(3);
+		final AtomicInteger replies = new AtomicInteger();
+		StopWatch stopWatch = new StopWatch();
+		stopWatch.start();
+		for (int i = 0; i < 3; i++) {
+			this.exchange(latch, gateway, replies);
+		}
+		latch.await();
+		stopWatch.stop();
+		assertTrue(stopWatch.getTotalTimeMillis() <= 11000);
+		assertEquals(1, replies.get());
+	}
+
+
+	private void exchange(final CountDownLatch latch, final RequestReplyExchanger gateway, final AtomicInteger replies) {
+		new Thread(new Runnable() {
+			public void run() {
+				try {
+					gateway.exchange(new GenericMessage<String>(""));
+					replies.incrementAndGet();
+				} catch (Exception e) {
+					//ignore
+				}
+				latch.countDown();
+			}
+		}).start();
+	}
+}

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java

@@ -0,0 +1,168 @@
+/*
+ * Copyright 2002-2012 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.jms.request_reply;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.MessageTimeoutException;
+import org.springframework.integration.gateway.RequestReplyExchanger;
+import org.springframework.integration.jms.config.ActiveMqTestUtils;
+import org.springframework.integration.message.GenericMessage;
+/**
+ * @author Oleg Zhurakousky
+ */
+public class PipelineJmsTests {
+
+	private final Executor executor = Executors.newFixedThreadPool(30);
+
+	int requests = 50;
+
+	/**
+	 * jms:out -> jms:in -> randomTimeoutProcess ->
+	 * jms:out -> jms:in
+	 * All reply queues are TEMPORARY
+	 */
+	@Test
+	public void testPipeline1() throws Exception{
+		this.test("pipeline-01.xml");
+	}
+
+	/**
+	 * jms:out(correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess ->
+	 * jms:out -> jms:in
+	 * All reply queues are TEMPORARY
+	 */
+	@Test
+	public void testPipeline2() throws Exception{
+		this.test("pipeline-02.xml");
+	}
+
+	/**
+	 * jms:out(correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess ->
+	 * jms:out(correlation-key="JMSCorrelationID") -> jms:in
+	 * All reply queues are TEMPORARY
+	 */
+	@Test
+	public void testPipeline3() throws Exception{
+		this.test("pipeline-03.xml");
+	}
+
+	/**
+	 * jms:out -> jms:in -> randomTimeoutProcess ->
+	 * jms:out(correlation-key="JMSCorrelationID") -> jms:in
+	 * All reply queues are TEMPORARY
+	 */
+	@Test
+	public void testPipeline4() throws Exception{
+		this.test("pipeline-04.xml");
+	}
+
+	/**
+	 * jms:out(correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess ->
+	 * jms:out -> jms:in
+	 * All reply queues are TEMPORARY
+	 */
+	@Test
+	public void testPipeline5() throws Exception{
+		this.test("pipeline-05.xml");
+	}
+
+	/**
+	 * jms:out -> jms:in -> randomTimeoutProcess ->
+	 * jms:out(correlation-key="foo") -> jms:in(correlation-key="foo")
+	 * All reply queues are TEMPORARY
+	 */
+	@Test
+	public void testPipeline6() throws Exception{
+		this.test("pipeline-06.xml");
+	}
+
+	/**
+	 * jms:out(correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess ->
+	 * jms:out(correlation-key="foo") -> jms:in(correlation-key="foo")
+	 * All reply queues are TEMPORARY
+	 */
+	@Test
+	public void testPipeline7() throws Exception{
+		this.test("pipeline-07.xml");
+	}
+
+	/**
+	 * jms:out(correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess ->
+	 * jms:out(correlation-key="JMSCorrelationID") -> jms:in
+	 * All reply queues are TEMPORARY
+	 */
+	@Test
+	public void testPipeline8() throws Exception{
+		this.test("pipeline-08.xml");
+	}
+
+	/**
+	 * jms:out(correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess ->
+	 * jms:out(correlation-key="bar") -> jms:in(correlation-key="bar")
+	 * All reply queues are TEMPORARY
+	 */
+	@Test
+	public void testPipeline9() throws Exception{
+		this.test("pipeline-09.xml");
+	}
+
+	public void test(String contextConfig) throws Exception{
+		ActiveMqTestUtils.prepare();
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(contextConfig, this.getClass());
+		final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
+		final CountDownLatch latch = new CountDownLatch(requests);
+		final AtomicInteger successCounter = new AtomicInteger();
+		final AtomicInteger timeoutCounter = new AtomicInteger();
+		final AtomicInteger failureCounter = new AtomicInteger();
+
+		for (int i = 0; i < requests; i++) {
+			final int y = i;
+			executor.execute(new Runnable() {
+				public void run() {
+					try {
+						assertEquals(y, gateway.exchange(new GenericMessage<Integer>(y)).getPayload());
+						successCounter.incrementAndGet();
+					} catch (MessageTimeoutException e) {
+						timeoutCounter.incrementAndGet();
+					} catch (Throwable t) {
+						failureCounter.incrementAndGet();
+					} finally {
+						latch.countDown();
+					}
+				}
+			});
+		}
+		latch.await();
+		System.out.println("Success: " + successCounter.get());
+		System.out.println("Timeout: " + timeoutCounter.get());
+		System.out.println("Failure: " + failureCounter.get());
+		// technically all we care that its > 0,
+		// but reality of this test it has to be something more then 0
+		assertTrue(successCounter.get() > 10);
+		assertEquals(0, failureCounter.get());
+		assertEquals(requests, successCounter.get() + timeoutCounter.get());
+		context.destroy();
+	}
+}

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java

@@ -0,0 +1,181 @@
+/*
+ * Copyright 2002-2012 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.jms.request_reply;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.MessageTimeoutException;
+import org.springframework.integration.gateway.RequestReplyExchanger;
+import org.springframework.integration.jms.config.ActiveMqTestUtils;
+import org.springframework.integration.message.GenericMessage;
+/**
+ * @author Oleg Zhurakousky
+ * @author Gary Russell
+ */
+public class PipelineNamedReplyQueuesJmsTests {
+
+	private final Executor executor = Executors.newFixedThreadPool(30);
+
+	int requests = 50;
+
+	int timeouts;
+
+	/**
+	 * jms:out(reply-destination-name="pipeline01-01") -> jms:in -> randomTimeoutProcess ->
+	 * jms:out -> jms:in
+	 */
+	@Test
+	public void testPipeline1() throws Exception{
+		this.test("pipeline-named-queue-01.xml");
+	}
+
+	/**
+	 * jms:out(reply-destination-name="pipeline02-01") -> jms:in -> randomTimeoutProcess ->
+	 * jms:out(reply-destination-name="pipeline02-02") -> jms:in ->
+	 * jms:out(reply-destination-name="pipeline02-03") -> jms:in
+	 */
+	@Test
+	public void testPipeline2() throws Exception{
+		this.test("pipeline-named-queue-02.xml");
+	}
+
+	/**
+	 * Same as {@link #testPipeline2()} except all gateways use the same reply queue.
+	 * and zero failures expected (no timeouts on server).
+	 * jms:out(reply-destination-name="pipeline02a-01") -> jms:in -> zeroTimeoutProcess ->
+	 * jms:out(reply-destination-name="pipeline02a-01") -> jms:in ->
+	 * jms:out(reply-destination-name="pipeline02a-01") -> jms:in
+	 */
+	@Test
+	public void testPipeline2a() throws Exception{
+		this.test("pipeline-named-queue-02a.xml");
+		assertEquals(0, this.timeouts);
+	}
+
+	/**
+	 * jms:out(reply-destination-name="pipeline03-01", correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess ->
+	 * jms:out(reply-destination-name="pipeline03-02") -> jms:in ->
+	 * jms:out(reply-destination-name="pipeline03-03") -> jms:in
+	 */
+	@Test
+	public void testPipeline3() throws Exception{
+		this.test("pipeline-named-queue-03.xml");
+	}
+
+	/**
+	 * Same as {@link #testPipeline3()} except all gateways use the same reply queue.
+	 * Ensures the correlation id is not propagated. No timeouts expected.
+	 * jms:out(reply-destination-name="pipeline03a-01", correlation-key="JMSCorrelationID") -> jms:in -> zeroTimeoutProcess ->
+	 * jms:out(reply-destination-name="pipeline03a-01") -> jms:in ->
+	 * jms:out(reply-destination-name="pipeline03a-01") -> jms:in
+	 * Ensures reply came from service after third gateway
+	 */
+	@Test
+	public void testPipeline3a() throws Exception{
+		this.test("pipeline-named-queue-03a.xml", 20000);
+		assertEquals(0, this.timeouts);
+	}
+
+	/**
+	 * jms:out(reply-destination-name="pipeline04-01", correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess ->
+	 * jms:out(reply-destination-name="pipeline04-02") -> jms:in
+	 */
+	@Test
+	public void testPipeline4() throws Exception{
+		this.test("pipeline-named-queue-04.xml");
+	}
+
+	/**
+	 * jms:out(reply-destination-name="pipeline05-01", correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess ->
+	 * jms:out(reply-destination-name="pipeline05-02", correlation-key="JMSCorrelationID") -> jms:in
+	 */
+	@Test
+	public void testPipeline5() throws Exception{
+		this.test("pipeline-named-queue-05.xml");
+	}
+
+	/**
+	 * jms:out(reply-destination-name="pipeline06-01", correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess ->
+	 * jms:out(reply-destination-name="pipeline06-02", correlation-key="foo") -> jms:in(correlation-key="foo")
+	 */
+	@Test
+	public void testPipeline6() throws Exception{
+		this.test("pipeline-named-queue-06.xml");
+	}
+
+	/**
+	 * jms:out(reply-destination-name="pipeline07-01", correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess ->
+	 * jms:out(reply-destination-name="pipeline07-02", correlation-key="foo") -> jms:in(correlation-key="foo")
+	 */
+	@Test
+	public void testPipeline7() throws Exception{
+		this.test("pipeline-named-queue-07.xml");
+	}
+
+	public void test(String contextConfig) throws Exception {
+		test(contextConfig, 0);
+	}
+
+	public void test(String contextConfig, final int offset) throws Exception {
+		this.timeouts = 0;
+		ActiveMqTestUtils.prepare();
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(contextConfig, this.getClass());
+		final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
+		final CountDownLatch latch = new CountDownLatch(requests);
+		final AtomicInteger successCounter = new AtomicInteger();
+		final AtomicInteger timeoutCounter = new AtomicInteger();
+		final AtomicInteger failureCounter = new AtomicInteger();
+
+		for (int i = 0; i < requests; i++) {
+			final int y = i;
+			executor.execute(new Runnable() {
+				public void run() {
+					try {
+						assertEquals(y + offset, gateway.exchange(new GenericMessage<Integer>(y)).getPayload());
+						successCounter.incrementAndGet();
+					} catch (MessageTimeoutException e) {
+						timeoutCounter.incrementAndGet();
+					} catch (Throwable t) {
+						t.printStackTrace();
+						failureCounter.incrementAndGet();
+					} finally {
+						latch.countDown();
+					}
+				}
+			});
+		}
+		assertTrue(latch.await(60, TimeUnit.SECONDS));
+		System.out.println("Success: " + successCounter.get());
+		System.out.println("Timeout: " + timeoutCounter.get());
+		System.out.println("Failure: " + failureCounter.get());
+		// technically all we care that its > 0,
+		// but reality of this test it has to be something more then 0
+		assertTrue(successCounter.get() > 10);
+		assertEquals(0, failureCounter.get());
+		assertEquals(requests, successCounter.get() + timeoutCounter.get());
+		this.timeouts = timeoutCounter.get();
+		context.destroy();
+	}
+}

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCachedConsumersTests.java

@@ -0,0 +1,246 @@
+/*
+ * Copyright 2002-2012 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.jms.request_reply;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.util.concurrent.CountDownLatch;
+
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.junit.Test;
+
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.MessageTimeoutException;
+import org.springframework.integration.gateway.RequestReplyExchanger;
+import org.springframework.integration.jms.JmsOutboundGateway;
+import org.springframework.integration.jms.config.ActiveMqTestUtils;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.jms.connection.CachingConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.core.MessageCreator;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
+import org.springframework.jms.listener.SessionAwareMessageListener;
+import org.springframework.jms.support.converter.SimpleMessageConverter;
+/**
+ * @author Oleg Zhurakousky
+ */
+public class RequestReplyScenariosWithCachedConsumersTests {
+
+	private final SimpleMessageConverter converter = new SimpleMessageConverter();
+
+	@Test(expected=MessageTimeoutException.class)
+	public void messageCorrelationBasedOnRequestMessageIdOptimized() throws Exception{
+		ActiveMqTestUtils.prepare();
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean("standardMessageIdCopyingConsumerWithOptimization", RequestReplyExchanger.class);
+		CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
+		final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
+
+
+		final Destination requestDestination = context.getBean("siOutQueueOptimizedA", Destination.class);
+		final Destination replyDestination = context.getBean("siInQueueOptimizedA", Destination.class);
+		new Thread(new Runnable() {
+
+			public void run() {
+				final Message requestMessage = jmsTemplate.receive(requestDestination);
+				jmsTemplate.send(replyDestination, new MessageCreator() {
+
+					public Message createMessage(Session session) throws JMSException {
+						TextMessage message = session.createTextMessage();
+						message.setText("bar");
+						message.setJMSCorrelationID(requestMessage.getJMSMessageID());
+						return message;
+					}
+				});
+			}
+		}).start();
+		gateway.exchange(new GenericMessage<String>("foo"));
+		context.close();
+	}
+
+	@Test
+	public void messageCorrelationBasedOnRequestMessageIdNonOptimized() throws Exception{
+		ActiveMqTestUtils.prepare();
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean("standardMessageIdCopyingConsumerWithoutOptimization", RequestReplyExchanger.class);
+		CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
+		final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
+
+
+		final Destination requestDestination = context.getBean("siOutQueueNonOptimizedB", Destination.class);
+		final Destination replyDestination = context.getBean("siInQueueNonOptimizedB", Destination.class);
+		new Thread(new Runnable() {
+
+			public void run() {
+				final Message requestMessage = jmsTemplate.receive(requestDestination);
+				jmsTemplate.send(replyDestination, new MessageCreator() {
+
+					public Message createMessage(Session session) throws JMSException {
+						TextMessage message = session.createTextMessage();
+						message.setText("bar");
+						message.setJMSCorrelationID(requestMessage.getJMSMessageID());
+						return message;
+					}
+				});
+			}
+		}).start();
+		org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
+		assertEquals("bar", siReplyMessage.getPayload());
+		context.close();
+	}
+
+	@Test
+	public void messageCorrelationBasedOnRequestCorrelationIdOptimized() throws Exception{
+		ActiveMqTestUtils.prepare();
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean("correlationPropagatingConsumerWithOptimization", RequestReplyExchanger.class);
+		CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
+
+		final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
+
+		final Destination requestDestination = context.getBean("siOutQueueOptimizedC", Destination.class);
+		final Destination replyDestination = context.getBean("siInQueueOptimizedC", Destination.class);
+		new Thread(new Runnable() {
+
+			public void run() {
+				final Message requestMessage = jmsTemplate.receive(requestDestination);
+				jmsTemplate.send(replyDestination, new MessageCreator() {
+
+					public Message createMessage(Session session) throws JMSException {
+						TextMessage message = session.createTextMessage();
+						message.setText("bar");
+						message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
+						return message;
+					}
+				});
+			}
+		}).start();
+		org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
+		assertEquals("bar", siReplyMessage.getPayload());
+		context.close();
+	}
+
+	@Test(expected=MessageTimeoutException.class)
+	public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() throws Exception{
+		ActiveMqTestUtils.prepare();
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean("correlationPropagatingConsumerWithoutOptimization", RequestReplyExchanger.class);
+		CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
+
+		final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
+
+		final Destination requestDestination = context.getBean("siOutQueueNonOptimizedD", Destination.class);
+		final Destination replyDestination = context.getBean("siInQueueNonOptimizedD", Destination.class);
+		new Thread(new Runnable() {
+
+			public void run() {
+				final Message requestMessage = jmsTemplate.receive(requestDestination);
+				jmsTemplate.send(replyDestination, new MessageCreator() {
+
+					public Message createMessage(Session session) throws JMSException {
+						TextMessage message = session.createTextMessage();
+						message.setText("bar");
+						message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
+						return message;
+					}
+				});
+			}
+		}).start();
+		org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
+		assertEquals("bar", siReplyMessage.getPayload());
+		context.close();
+	}
+
+	@Test
+	public void messageCorrelationBasedOnRequestCorrelationIdTimedOutFirstReplyOptimized() throws Exception{
+		ActiveMqTestUtils.prepare();
+		ClassPathXmlApplicationContext context =
+				new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
+		RequestReplyExchanger gateway =
+				context.getBean("correlationPropagatingConsumerWithOptimizationDelayFirstReply", RequestReplyExchanger.class);
+		final ConnectionFactory connectionFactory = context.getBean("connectionFactory", ConnectionFactory.class);
+
+		final Destination requestDestination = context.getBean("siOutQueueE", Destination.class);
+		final Destination replyDestination = context.getBean("siInQueueE", Destination.class);
+
+		for (int i = 0; i < 3; i++) {
+			System.out.println("#### " + i);
+			try {
+				gateway.exchange(gateway.exchange(new GenericMessage<String>("foo")));
+			} catch (Exception e) {/*ignore*/}
+
+		}
+
+		final CountDownLatch latch = new CountDownLatch(1);
+		new Thread(new Runnable() {
+
+			public void run() {
+				DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
+				dmlc.setConnectionFactory(connectionFactory);
+				dmlc.setDestination(requestDestination);
+				dmlc.setMessageListener(new SessionAwareMessageListener<Message>() {
+
+					public void onMessage(Message message, Session session) {
+						String requestPayload = (String) extractPayload(message);
+						try {
+							TextMessage replyMessage = session.createTextMessage();
+							replyMessage.setText(requestPayload);
+							replyMessage.setJMSCorrelationID(message.getJMSCorrelationID());
+							MessageProducer producer = session.createProducer(replyDestination);
+							producer.send(replyMessage);
+						} catch (Exception e) {
+							// ignore. the test will fail
+						}
+					}
+				});
+				dmlc.afterPropertiesSet();
+				dmlc.start();
+				latch.countDown();
+			}
+		}).start();
+
+		latch.await();
+
+
+
+		TestUtils.getPropertyValue(context.getBean("fastGateway"), "handler", JmsOutboundGateway.class).setReceiveTimeout(10000);
+		Thread.sleep(1000);
+		assertEquals("bar", gateway.exchange(new GenericMessage<String>("bar")).getPayload());
+
+		context.close();
+	}
+
+	private Object extractPayload(Message jmsMessage) {
+		try {
+			return converter.fromMessage(jmsMessage);
+		} catch (Exception e) {
+			e.printStackTrace();
+			fail();
+		}
+		return null;
+	}
+}

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2002-2012 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.jms.request_reply;
+
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.gateway.RequestReplyExchanger;
+import org.springframework.integration.jms.JmsOutboundGateway;
+import org.springframework.integration.jms.config.ActiveMqTestUtils;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
+/**
+ * @author Oleg Zhurakousky
+ */
+public class RequestReplyScenariosWithCorrelationKeyProvidedTests {
+
+	@Test
+	public void messageCorrelationBasedCustomCorrelationKey() throws Exception{
+		ActiveMqTestUtils.prepare();
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean("explicitCorrelationKeyGateway", RequestReplyExchanger.class);
+
+		gateway.exchange(MessageBuilder.withPayload("foo").build());
+		context.close();
+	}
+
+	@Test
+	public void messageCorrelationBasedCustomCorrelationKeyAsJMSCorrelationID() throws Exception{
+		ActiveMqTestUtils.prepare();
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean("explicitCorrelationKeyGatewayB", RequestReplyExchanger.class);
+
+		gateway.exchange(MessageBuilder.withPayload("foo").build());
+		context.close();
+	}
+
+	@Test
+	public void messageCorrelationBasedCustomCorrelationKeyDelayedReplies() throws Exception{
+		ActiveMqTestUtils.prepare();
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean("explicitCorrelationKeyGatewayC", RequestReplyExchanger.class);
+
+
+		for (int i = 0; i < 3; i++) {
+			try {
+				gateway.exchange(MessageBuilder.withPayload("hello").build());
+			} catch (Exception e) {
+				// ignore
+			}
+		}
+
+		JmsOutboundGateway outGateway = TestUtils.getPropertyValue(context.getBean("outGateway"), "handler", JmsOutboundGateway.class);
+		outGateway.setReceiveTimeout(5000);
+		assertEquals("foo", gateway.exchange(MessageBuilder.withPayload("foo").build()).getPayload());
+		context.close();
+	}
+
+
+	public static class DelayedService {
+		public String echo(String s) throws Exception{
+			Thread.sleep(200);
+			return s;
+		}
+	}
+}

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithNonCachedConsumersTests.java

@@ -0,0 +1,157 @@
+/*
+ * Copyright 2002-2012 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.jms.request_reply;
+
+import static org.junit.Assert.assertEquals;
+
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.junit.Test;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.MessageTimeoutException;
+import org.springframework.integration.gateway.RequestReplyExchanger;
+import org.springframework.integration.jms.config.ActiveMqTestUtils;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.core.MessageCreator;
+/**
+ * @author Oleg Zhurakousky
+ */
+public class RequestReplyScenariosWithNonCachedConsumersTests {
+
+	@Test(expected=MessageTimeoutException.class)
+	public void messageCorrelationBasedOnRequestMessageIdOptimized() throws Exception{
+		ActiveMqTestUtils.prepare();
+		ApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean("optimizedMessageId", RequestReplyExchanger.class);
+		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
+		final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
+
+		final Destination requestDestination = context.getBean("siOutQueueC", Destination.class);
+		final Destination replyDestination = context.getBean("siInQueueC", Destination.class);
+		new Thread(new Runnable() {
+
+			public void run() {
+				final Message requestMessage = jmsTemplate.receive(requestDestination);
+				jmsTemplate.send(replyDestination, new MessageCreator() {
+
+					public Message createMessage(Session session) throws JMSException {
+						TextMessage message = session.createTextMessage();
+						message.setText("bar");
+						message.setJMSCorrelationID(requestMessage.getJMSMessageID());
+						return message;
+					}
+				});
+			}
+		}).start();
+		org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
+		assertEquals("bar", siReplyMessage.getPayload());
+	}
+
+	@Test
+	public void messageCorrelationBasedOnRequestMessageIdNonOptimized() throws Exception{
+		ActiveMqTestUtils.prepare();
+		ApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean("nonoptimizedMessageId", RequestReplyExchanger.class);
+		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
+		final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
+
+		final Destination requestDestination = context.getBean("siOutQueueD", Destination.class);
+		final Destination replyDestination = context.getBean("siInQueueD", Destination.class);
+		new Thread(new Runnable() {
+
+			public void run() {
+				final Message requestMessage = jmsTemplate.receive(requestDestination);
+				jmsTemplate.send(replyDestination, new MessageCreator() {
+
+					public Message createMessage(Session session) throws JMSException {
+						TextMessage message = session.createTextMessage();
+						message.setText("bar");
+						message.setJMSCorrelationID(requestMessage.getJMSMessageID());
+						return message;
+					}
+				});
+			}
+		}).start();
+		org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
+		assertEquals("bar", siReplyMessage.getPayload());
+	}
+
+	@Test
+	public void messageCorrelationBasedOnRequestCorrelationIdOptimized() throws Exception{
+		ActiveMqTestUtils.prepare();
+		ApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean("optimized", RequestReplyExchanger.class);
+		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
+		final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
+
+		final Destination requestDestination = context.getBean("siOutQueueA", Destination.class);
+		final Destination replyDestination = context.getBean("siInQueueA", Destination.class);
+		new Thread(new Runnable() {
+
+			public void run() {
+				final Message requestMessage = jmsTemplate.receive(requestDestination);
+				jmsTemplate.send(replyDestination, new MessageCreator() {
+
+					public Message createMessage(Session session) throws JMSException {
+						TextMessage message = session.createTextMessage();
+						message.setText("bar");
+						message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
+						return message;
+					}
+				});
+			}
+		}).start();
+		org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
+		assertEquals("bar", siReplyMessage.getPayload());
+	}
+
+	@Test(expected=MessageTimeoutException.class)
+	public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() throws Exception{
+		ActiveMqTestUtils.prepare();
+		ApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean("nonoptimized", RequestReplyExchanger.class);
+		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
+		final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
+
+		final Destination requestDestination = context.getBean("siOutQueueB", Destination.class);
+		final Destination replyDestination = context.getBean("siInQueueB", Destination.class);
+		new Thread(new Runnable() {
+
+			public void run() {
+				final Message requestMessage = jmsTemplate.receive(requestDestination);
+				jmsTemplate.send(replyDestination, new MessageCreator() {
+
+					public Message createMessage(Session session) throws JMSException {
+						TextMessage message = session.createTextMessage();
+						message.setText("bar");
+						message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
+						return message;
+					}
+				});
+			}
+		}).start();
+		org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
+		assertEquals("bar", siReplyMessage.getPayload());
+	}
+}

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java

@@ -0,0 +1,275 @@
+/*
+ * Copyright 2002-2012 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.jms.request_reply;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.junit.Test;
+
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.gateway.RequestReplyExchanger;
+import org.springframework.integration.jms.config.ActiveMqTestUtils;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.test.util.TestUtils;
+import org.springframework.jms.connection.CachingConnectionFactory;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.core.MessageCreator;
+import org.springframework.jms.listener.DefaultMessageListenerContainer;
+import org.springframework.jms.listener.SessionAwareMessageListener;
+import org.springframework.jms.support.converter.SimpleMessageConverter;
+/**
+ * @author Oleg Zhurakousky
+ */
+public class RequestReplyScenariosWithTempReplyQueuesTests {
+
+	private final SimpleMessageConverter converter = new SimpleMessageConverter();
+
+	@Test
+	public void messageCorrelationBasedOnRequestMessageId() throws Exception{
+		ActiveMqTestUtils.prepare();
+
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
+		CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
+		final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
+
+		final Destination requestDestination = context.getBean("siOutQueue", Destination.class);
+
+		new Thread(new Runnable() {
+
+			public void run() {
+				final Message requestMessage = jmsTemplate.receive(requestDestination);
+				Destination replyTo = null;
+				try {
+					replyTo = requestMessage.getJMSReplyTo();
+				} catch (Exception e) {
+					fail();
+				}
+				jmsTemplate.send(replyTo, new MessageCreator() {
+
+					public Message createMessage(Session session) throws JMSException {
+						try {
+							TextMessage message = session.createTextMessage();
+							message.setText("bar");
+							message.setJMSCorrelationID(requestMessage.getJMSMessageID());
+							return message;
+						} catch (Exception e) {
+							// ignore
+						}
+						return null;
+					}
+				});
+			}
+		}).start();
+		gateway.exchange(new GenericMessage<String>("foo"));
+		context.close();
+	}
+
+	@Test
+	public void messageCorrelationBasedOnRequestCorrelationIdTimedOutFirstReply() throws Exception{
+		ActiveMqTestUtils.prepare();
+		ClassPathXmlApplicationContext context =
+				new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", this.getClass());
+		RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
+		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
+
+		final Destination requestDestination = context.getBean("siOutQueue", Destination.class);
+
+		DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
+		dmlc.setConnectionFactory(connectionFactory);
+		dmlc.setDestination(requestDestination);
+		dmlc.setMessageListener(new SessionAwareMessageListener<Message>() {
+
+			public void onMessage(Message message, Session session) {
+				Destination replyTo = null;
+				try {
+					replyTo = message.getJMSReplyTo();
+				} catch (Exception e) {
+					fail();
+				}
+				String requestPayload = (String) extractPayload(message);
+				if (requestPayload.equals("foo")){
+					try {
+						Thread.sleep(6000);
+					} catch (Exception e) {/*ignore*/}
+				}
+				try {
+					TextMessage replyMessage = session.createTextMessage();
+					replyMessage.setText(requestPayload);
+					replyMessage.setJMSCorrelationID(message.getJMSMessageID());
+					MessageProducer producer = session.createProducer(replyTo);
+					producer.send(replyMessage);
+				} catch (Exception e) {
+					// ignore. the test will fail
+				}
+			}
+		});
+		dmlc.afterPropertiesSet();
+		dmlc.start();
+
+		try {
+			gateway.exchange(new GenericMessage<String>("foo"));
+		} catch (Exception e) {
+			// ignore
+		}
+		Thread.sleep(1000);
+		try {
+			assertEquals("bar", gateway.exchange(new GenericMessage<String>("bar")).getPayload());
+		} catch (Exception e) {
+			e.printStackTrace();
+			fail();
+		}
+		context.close();
+	}
+
+	/**
+	 * Validates that JOG will recreate a temporary queue
+	 * once a failure detected and that the messages will still be properly correlated
+	 */
+	@Test
+	public void brokenBrokerTest() throws Exception{
+
+		BrokerService broker = new BrokerService();
+		broker.setPersistent(false);
+		broker.setUseJmx(false);
+		broker.setTransportConnectorURIs(new String[]{"tcp://localhost:61623"});
+		broker.setDeleteAllMessagesOnStartup(true);
+		broker.start();
+
+		ClassPathXmlApplicationContext context =
+				new ClassPathXmlApplicationContext("broken-broker.xml", this.getClass());
+
+		final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
+
+		int replyCounter = 0;
+		int timeoutCounter = 0;
+		for (int i = 0; i < 50; i++) {
+			try {
+				assertEquals(i+"", gateway.exchange(new GenericMessage<String>(String.valueOf(i))).getPayload());
+				replyCounter++;
+			} catch (Exception e) {
+				timeoutCounter++;
+			}
+			if (i == 0 || i == 20 || i == 40){
+				Object replyDestination = TestUtils.getPropertyValue(context.getBean("jog"), "handler.replyDestination");
+				if (replyDestination != null){
+					broker.removeDestination((ActiveMQDestination) replyDestination);
+				}
+			}
+		}
+		assertEquals(50, replyCounter + timeoutCounter);
+	}
+
+	@Test
+	public void testConcurrently() throws Exception{
+		ActiveMqTestUtils.prepare();
+		ClassPathXmlApplicationContext context =
+				new ClassPathXmlApplicationContext("mult-producer-and-consumers-temp-reply.xml", this.getClass());
+		final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
+		Executor executor = Executors.newFixedThreadPool(10);
+		final int testNumbers = 100;
+		final CountDownLatch latch = new CountDownLatch(testNumbers);
+		final AtomicInteger failures = new AtomicInteger();
+		final AtomicInteger timeouts = new AtomicInteger();
+		final AtomicInteger missmatches = new AtomicInteger();
+		for (int i = 0; i < testNumbers; i++) {
+			final int y = i;
+			executor.execute(new Runnable() {
+				public void run() {
+					try {
+
+						String reply = (String) gateway.exchange(new GenericMessage<String>(String.valueOf(y))).getPayload();
+						if (!String.valueOf(y).equals(reply)){
+							missmatches.incrementAndGet();
+						}
+					} catch (Exception e) {
+						if (e instanceof MessageDeliveryException) {
+							timeouts.incrementAndGet();
+						}
+						else {
+							failures.incrementAndGet();
+						}
+					}
+//					if (latch.getCount()%100 == 0){
+//						long count = testNumbers-latch.getCount();
+//						if (count > 0){
+//							print(failures, timeouts, missmatches, testNumbers-latch.getCount());
+//						}
+//					}
+					latch.countDown();
+				}
+			});
+		}
+		latch.await();
+		print(failures, timeouts, missmatches, testNumbers);
+		Thread.sleep(5000);
+		assertEquals(0, missmatches.get());
+		assertEquals(0, failures.get());
+		assertEquals(0, timeouts.get());
+	}
+
+	private void print(AtomicInteger failures, AtomicInteger timeouts, AtomicInteger missmatches, long echangesProcessed){
+		System.out.println("============================");
+		System.out.println(echangesProcessed + " exchanges processed");
+		System.out.println("Failures: " + failures.get());
+		System.out.println("Timeouts: " + timeouts.get());
+		System.out.println("Missmatches: " + missmatches.get());
+		System.out.println("============================");
+	}
+
+	public static class MyRandomlySlowService{
+		Random random = new Random();
+		List<Integer> list = new ArrayList<Integer>();
+		public String secho(String value) throws Exception{
+			int i = random.nextInt(2000);
+//			if (i >= 2000){
+//				System.out.println("SLEEPIING: " + i);
+//			}
+			Thread.sleep(i);
+			return value;
+		}
+	}
+
+	private Object extractPayload(Message jmsMessage) {
+		try {
+			return converter.fromMessage(jmsMessage);
+		} catch (Exception e) {
+			e.printStackTrace();
+			fail();
+		}
+		return null;
+	}
+}

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/broken-broker.xml

@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.2.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<int:gateway id="brkenBrokerGateway" default-request-channel="outGatewayInChannel"/>
+	
+	<int-jms:outbound-gateway id="jog" request-channel="outGatewayInChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="brokenBrokerRequestQueue"
+		correlation-key="JMSCorrelationID"
+		receive-timeout="1000"/>
+		
+	<int-jms:inbound-gateway request-channel="jmsInChannel" 
+						     request-destination-name="brokenBrokerRequestQueue"
+							 connection-factory="connectionFactory"
+							 concurrent-consumers="10"
+							 reply-timeout="10000"/>
+							 
+	<int:channel id="jmsInChannel">
+		<int:dispatcher task-executor="executor"/>
+	</int:channel>
+							 		 
+	<int:service-activator input-channel="jmsInChannel" expression="payload"/>
+		
+	<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="tcp://localhost:61623"/>		
+			</bean>
+		</property>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+		<property name="sessionCacheSize" value="10" />
+	</bean>
+	
+	<task:executor id="executor" pool-size="20"/>
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml

@@ -0,0 +1,82 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.2.xsd">
+
+	<int:gateway id="explicitCorrelationKeyGateway" default-request-channel="explicitCorrelationIn"/>
+
+	<int-jms:outbound-gateway request-channel="explicitCorrelationIn"
+	    connection-factory="connectionFactory" 
+		request-destination="explicitCorrelationJmsOut"
+		correlation-key="bar"/>
+		
+	<bean id="explicitCorrelationJmsOut" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="explicitCorrelationJmsOut"/>
+	</bean>
+	
+	<int-jms:inbound-gateway request-channel="requestIn" 
+	    request-destination="explicitCorrelationJmsOut"
+	    correlation-key="bar"
+		connection-factory="connectionFactory"/>
+	
+	<int:transformer input-channel="requestIn" expression="payload"/>
+	
+	<!--  -->
+	
+	<int:gateway id="explicitCorrelationKeyGatewayB" default-request-channel="explicitCorrelationInB"/>
+
+	<int-jms:outbound-gateway request-channel="explicitCorrelationInB"
+	    connection-factory="connectionFactory" 
+		request-destination="explicitCorrelationJmsOutB"
+		correlation-key="JMSCorrelationID"/>
+		
+	<bean id="explicitCorrelationJmsOutB" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="explicitCorrelationJmsOutB"/>
+	</bean>
+	
+	<int-jms:inbound-gateway request-channel="requestInB" 
+	    request-destination="explicitCorrelationJmsOutB"
+	    correlation-key="JMSCorrelationID"
+		connection-factory="connectionFactory"/>
+	
+	<int:transformer input-channel="requestInB" expression="payload"/>
+	
+	<!--  -->
+	
+	<int:gateway id="explicitCorrelationKeyGatewayC" default-request-channel="explicitCorrelationInC"/>
+
+	<int-jms:outbound-gateway id="outGateway" request-channel="explicitCorrelationInC"
+	    connection-factory="connectionFactory" 
+		request-destination="explicitCorrelationJmsOutC"
+		reply-destination-name="explicitCorrelationJmsInC"
+		correlation-key="foo"
+		receive-timeout="100"/>
+		
+	<bean id="explicitCorrelationJmsOutC" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="explicitCorrelationJmsOutC"/>
+	</bean>
+	
+	<int-jms:inbound-gateway id="inGateway" request-channel="requestInC" 
+	    request-destination="explicitCorrelationJmsOutC"
+	    correlation-key="foo"
+		connection-factory="connectionFactory"/>
+	
+	<int:transformer input-channel="requestInC">
+		<bean class="org.springframework.integration.jms.request_reply.RequestReplyScenariosWithCorrelationKeyProvidedTests.DelayedService"/>
+	</int:transformer>
+	
+	<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>		
+			</bean>
+		</property>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+		<property name="sessionCacheSize" value="10" />
+	</bean>
+	
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/honor-timeout.xml

@@ -0,0 +1,44 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="in" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="in"
+	    connection-factory="connectionFactory" 
+		request-destination-name="honorTimeoutQueue"
+		receive-timeout="10000"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="honorTimeoutQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="1"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn">
+		<int:header-enricher>
+			<int:header name="delay" expression="9000"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/mult-producer-and-consumers-temp-reply.xml

@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.2.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
+
+	<int:gateway id="multiOutGateway" default-request-channel="outGatewayInChannel"/>
+	
+	<int:channel id="outGatewayInChannel">
+		<int:dispatcher task-executor="executor"/>
+	</int:channel>
+	
+	<int-jms:outbound-gateway request-channel="outGatewayInChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="multiOutGatewayTempQueue"
+		correlation-key="JMSCorrelationID"/>
+		
+	<int-jms:inbound-gateway request-channel="jmsInChannel" 
+						     request-destination-name="multiOutGatewayTempQueue"
+							 connection-factory="connectionFactory"
+							 concurrent-consumers="10"
+							 reply-timeout="10000"/>
+							 
+	<int:channel id="jmsInChannel">
+		<int:dispatcher task-executor="executor"/>
+	</int:channel>
+							 
+	<int:service-activator input-channel="jmsInChannel">
+		<bean class="org.springframework.integration.jms.request_reply.RequestReplyScenariosWithTempReplyQueuesTests.MyRandomlySlowService"/>
+	</int:service-activator>
+		
+	<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>		
+			</bean>
+		</property>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+		<property name="sessionCacheSize" value="10" />
+	</bean>
+	
+	<task:executor id="executor" pool-size="20"/>
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-01.xml

@@ -0,0 +1,58 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline01" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline01"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline01-queue-01"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="pipeline01-queue-01"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline01-queue-02"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="pipeline01-queue-02"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-02.xml

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline02" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline02"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline02-queue-01"
+		correlation-key="JMSCorrelationID"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="pipeline02-queue-01"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline02-queue-02"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="pipeline02-queue-02"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-03.xml

@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline03" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline03"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline03-queue-01"
+		correlation-key="JMSCorrelationID"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="pipeline03-queue-01"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline03-queue-02"
+		correlation-key="JMSCorrelationID"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="pipeline03-queue-02"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-04.xml

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline04" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline04"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline04-queue-01"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="pipeline04-queue-01"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline04-queue-02"
+		correlation-key="JMSCorrelationID"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="pipeline04-queue-02"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-05.xml

@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline05" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline05"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline05-queue-01"
+		correlation-key="foo"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="pipeline05-queue-01"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"
+		correlation-key="foo"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline05-queue-02"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="pipeline05-queue-02"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-06.xml

@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline06" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline06"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline06-queue-01"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="pipeline06-queue-01"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline06-queue-02"
+		correlation-key="foo"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="pipeline06-queue-02"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"
+		correlation-key="foo"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-07.xml

@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline07" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline07"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline07-queue-01"
+		correlation-key="JMSCorrelationID"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="pipeline07-queue-01"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline07-queue-02"
+		correlation-key="foo"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="pipeline07-queue-02"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"
+		correlation-key="foo"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-08.xml

@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline08" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline08"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline08-queue-01"
+		correlation-key="foo"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="pipeline08-queue-01"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"
+		correlation-key="foo"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline08-queue-02"
+		correlation-key="JMSCorrelationID"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="pipeline08-queue-02"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-09.xml

@@ -0,0 +1,62 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline09" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline09"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline09-queue-01"
+		correlation-key="foo"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="pipeline09-queue-01"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"
+		correlation-key="foo"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="pipeline09-queue-02"
+		correlation-key="bar"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="pipeline09-queue-02"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"
+		correlation-key="bar"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-01.xml

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline01" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline01"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline01-01"
+		request-destination-name="siOutQueue"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="siOutQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+		request-destination-name="anotherGatewayQueue"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="anotherGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02.xml

@@ -0,0 +1,73 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline02" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline02"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline02-01"
+		request-destination-name="siOutQueue"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="siOutQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline02-02"
+		request-destination-name="anotherGatewayQueue"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="anotherGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload" output-channel="toThird" />
+
+	<int-jms:outbound-gateway request-channel="toThird"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline02-03"
+		request-destination-name="thirdGatewayQueue"/>
+		
+	<int-jms:inbound-gateway request-channel="thirdIn"
+		request-destination-name="thirdGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:bridge input-channel="thirdIn" />
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02a.xml

@@ -0,0 +1,73 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline02" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline02"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline02a-01"
+		request-destination-name="siOutQueue"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="siOutQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="0"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline02a-01"
+		request-destination-name="anotherGatewayQueue"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="anotherGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload" output-channel="toThird" />
+
+	<int-jms:outbound-gateway request-channel="toThird"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline02a-01"
+		request-destination-name="thirdGatewayQueue"/>
+		
+	<int-jms:inbound-gateway request-channel="thirdIn"
+		request-destination-name="thirdGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:bridge input-channel="thirdIn" />
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03.xml

@@ -0,0 +1,74 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline03" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline03"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline03-01"
+	    correlation-key="JMSCorrelationID"
+		request-destination-name="siOutQueue"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="siOutQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline03-02"
+		request-destination-name="anotherGatewayQueue"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="anotherGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload" output-channel="toThird" />
+
+	<int-jms:outbound-gateway request-channel="toThird"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline03-03"
+		request-destination-name="thirdGatewayQueue"/>
+		
+	<int-jms:inbound-gateway request-channel="thirdIn"
+		request-destination-name="thirdGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:bridge input-channel="thirdIn" />
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml

@@ -0,0 +1,77 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline03" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline03"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline03a-01"
+	    correlation-key="JMSCorrelationID"
+		request-destination-name="siOutQueue"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="siOutQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="0"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline03a-01"
+		request-destination-name="anotherGatewayQueue"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="anotherGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload" output-channel="toThird" />
+
+	<int-jms:outbound-gateway request-channel="toThird"
+		reply-channel="add10kOnTheWayBack"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline03a-01"
+		request-destination-name="thirdGatewayQueue"/>
+
+	<int:transformer input-channel="add10kOnTheWayBack" expression="payload + 10000" />
+		
+	<int-jms:inbound-gateway request-channel="thirdIn"
+		request-destination-name="thirdGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+
+	<int:transformer input-channel="thirdIn" expression="payload + 10000" />
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml

@@ -0,0 +1,62 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline04" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline04"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline04-01"
+	    correlation-key="foo"
+		request-destination-name="siOutQueue"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="siOutQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		correlation-key="foo"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline04-02"
+		request-destination-name="anotherGatewayQueue"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="anotherGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-05.xml

@@ -0,0 +1,63 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline05" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline05"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline05-01"
+	    correlation-key="foo"
+		request-destination-name="siOutQueue"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="siOutQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		correlation-key="foo"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline05-02"
+		request-destination-name="anotherGatewayQueue"
+		correlation-key="JMSCorrelationID"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="anotherGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-06.xml

@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline06" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline06"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline06-01"
+	    correlation-key="JMSCorrelationID"
+		request-destination-name="siOutQueue"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="siOutQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		correlation-key="JMSCorrelationID"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline06-02"
+		request-destination-name="anotherGatewayQueue"
+		correlation-key="foo"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="anotherGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"
+		correlation-key="foo"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-07.xml

@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:task="http://www.springframework.org/schema/task"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.1.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
+
+	<int:gateway default-request-channel="pipeline07" default-request-timeout="10000" default-reply-timeout="10000"/>
+
+	<int-jms:outbound-gateway request-channel="pipeline07"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline07-01"
+	    correlation-key="foo"
+		request-destination-name="siOutQueue"/>
+	
+	<int-jms:inbound-gateway request-channel="jmsIn"
+		request-destination-name="siOutQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		correlation-key="foo"
+		reply-timeout="10000"/>
+
+	<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
+		<int:header-enricher>
+			<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
+		</int:header-enricher>
+		<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
+		<int:transformer expression="payload"/>
+	</int:chain>
+	
+	<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
+	    connection-factory="connectionFactory" 
+	    reply-destination-name="pipeline07-02"
+		request-destination-name="anotherGatewayQueue"
+		correlation-key="bar"/>
+		
+	<int-jms:inbound-gateway request-channel="anotherIn"
+		request-destination-name="anotherGatewayQueue"
+		connection-factory="connectionFactory"
+		concurrent-consumers="10"
+		reply-timeout="10000"
+		correlation-key="bar"/>
+		
+	<int:transformer input-channel="anotherIn" expression="payload"/>
+
+	<bean id="connectionFactory"
+		class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>
+			</bean>
+		</property>
+		<property name="sessionCacheSize" value="10"/>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+	</bean>
+
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-cached-consumers.xml

@@ -0,0 +1,107 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.2.xsd">
+
+	<int:gateway id="standardMessageIdCopyingConsumerWithOptimization" default-request-channel="jmsInOptimizedA"/>
+
+	<int-jms:outbound-gateway request-channel="jmsInOptimizedA"
+	    connection-factory="connectionFactory" 
+		request-destination="siOutQueueOptimizedA"
+		reply-destination="siInQueueOptimizedA"
+		correlation-key="JMSCorrelationID"/>
+		
+	<bean id="siOutQueueOptimizedA" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siOutQueueOptimizedA"/>
+	</bean>
+	
+	<bean id="siInQueueOptimizedA" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siInQueueOptimizedA"/>
+	</bean>
+	
+	<!--  -->
+	
+	<int:gateway id="standardMessageIdCopyingConsumerWithoutOptimization" default-request-channel="jmsInNonOptimizedB"/>
+
+	<int-jms:outbound-gateway request-channel="jmsInNonOptimizedB"
+	    connection-factory="connectionFactory" 
+		request-destination="siOutQueueNonOptimizedB"
+		reply-destination="siInQueueNonOptimizedB"/>
+		
+	<bean id="siOutQueueNonOptimizedB" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siOutQueueNonOptimizedB"/>
+	</bean>
+	
+	<bean id="siInQueueNonOptimizedB" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siInQueueNonOptimizedB"/>
+	</bean>
+	
+	<!--  -->
+	
+	<int:gateway id="correlationPropagatingConsumerWithOptimization" default-request-channel="jmsInOptimizedC"/>
+
+	<int-jms:outbound-gateway request-channel="jmsInOptimizedC"
+	    connection-factory="connectionFactory" 
+		request-destination="siOutQueueOptimizedC"
+		reply-destination="siInQueueOptimizedC"
+		correlation-key="JMSCorrelationID"/>
+		
+	<bean id="siOutQueueOptimizedC" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siOutQueueOptimizedC"/>
+	</bean>
+	
+	<bean id="siInQueueOptimizedC" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siInQueueOptimizedC"/>
+	</bean>
+	
+	<!--  -->
+	
+	<int:gateway id="correlationPropagatingConsumerWithoutOptimization" default-request-channel="jmsInNonOptimizedD"/>
+
+	<int-jms:outbound-gateway request-channel="jmsInNonOptimizedD"
+	    connection-factory="connectionFactory" 
+		request-destination="siOutQueueNonOptimizedD"
+		reply-destination="siInQueueNonOptimizedD"/>
+		
+	<bean id="siOutQueueNonOptimizedD" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siOutQueueNonOptimizedD"/>
+	</bean>
+	
+	<bean id="siInQueueNonOptimizedD" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siInQueueNonOptimizedD"/>
+	</bean>
+	
+	<!--  -->
+	
+	<int:gateway id="correlationPropagatingConsumerWithOptimizationDelayFirstReply" default-request-channel="jmsInE"/>
+
+	<int-jms:outbound-gateway id="fastGateway" request-channel="jmsInE"
+	    connection-factory="connectionFactory" 
+		request-destination="siOutQueueE"
+		reply-destination="siInQueueE"
+		receive-timeout="500"
+		correlation-key="JMSCorrelationID"/>
+		
+	<bean id="siOutQueueE" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siOutQueueE"/>
+	</bean>
+	
+	<bean id="siInQueueE" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siInQueueE"/>
+	</bean>
+	 
+	<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>		
+			</bean>
+		</property>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+		<property name="sessionCacheSize" value="10" />
+	</bean>
+	
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-no-cached-consumers.xml

@@ -0,0 +1,87 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.2.xsd">
+
+	<int:gateway id="optimized" default-request-channel="jmsInOptimized"/>
+
+	<int-jms:outbound-gateway request-channel="jmsInOptimized"
+	    connection-factory="connectionFactory" 
+		request-destination="siOutQueueA"
+		reply-destination="siInQueueA"
+		correlation-key="JMSCorrelationID"/>
+		
+	<bean id="siOutQueueA" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siOutQueueA.not.cached"/>
+	</bean>
+	
+	<bean id="siInQueueA" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siInQueueA.not.cached"/>
+	</bean>
+	
+	<!--  -->
+	
+	<int:gateway id="nonoptimized" default-request-channel="jmsInNonOptimized"/>
+
+	<int-jms:outbound-gateway request-channel="jmsInNonOptimized"
+	    connection-factory="connectionFactory" 
+		request-destination="siOutQueueB"
+		reply-destination="siInQueueB"/>
+		
+	<bean id="siOutQueueB" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siOutQueueB.not.cached"/>
+	</bean>
+	
+	<bean id="siInQueueB" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siInQueueB.not.cached"/>
+	</bean>
+	
+	<!--  -->
+	
+	<int:gateway id="optimizedMessageId" default-request-channel="jmsInOptimizedC"/>
+
+	<int-jms:outbound-gateway request-channel="jmsInOptimizedC"
+	    connection-factory="connectionFactory" 
+		request-destination="siOutQueueC"
+		reply-destination="siInQueueC"
+		correlation-key="JMSCorrelationID"/>
+		
+	<bean id="siOutQueueC" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siOutQueueC.not.cached"/>
+	</bean>
+	
+	<bean id="siInQueueC" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siInQueueC.not.cached"/>
+	</bean>
+	
+	<!--  -->
+	
+	<int:gateway id="nonoptimizedMessageId" default-request-channel="jmsInNonOptimizedD"/>
+
+	<int-jms:outbound-gateway request-channel="jmsInNonOptimizedD"
+	    connection-factory="connectionFactory" 
+		request-destination="siOutQueueD"
+		reply-destination="siInQueueD"/>
+		
+	<bean id="siOutQueueD" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siOutQueueD.not.cached"/>
+	</bean>
+	
+	<bean id="siInQueueD" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siInQueueD.not.cached"/>
+	</bean>
+	 
+	<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>		
+			</bean>
+		</property>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="false" />
+		<property name="sessionCacheSize" value="10" />
+	</bean>
+</beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-temp-reply-consumers.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.2.xsd">
+
+	<int:gateway default-request-channel="jmsIn"/>
+
+	<int-jms:outbound-gateway request-channel="jmsIn"
+	    connection-factory="connectionFactory" 
+		request-destination="siOutQueue"/>
+		
+	<bean id="siOutQueue" class="org.apache.activemq.command.ActiveMQQueue">
+		<constructor-arg value="siOutQueueA"/>
+	</bean>
+	 
+	<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
+		<property name="targetConnectionFactory">
+			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
+				<property name="brokerURL" value="vm://localhost"/>		
+			</bean>
+		</property>
+		<property name="cacheProducers" value="true" />
+		<property name="cacheConsumers" value="true" />
+		<property name="sessionCacheSize" value="10" />
+	</bean>
+	
+</beans>

spring-integration-jms/src/test/resources/log4j.properties

@@ -1,11 +1,11 @@
-log4j.rootCategory=WARN, stdout
+log4j.rootCategory=ERROR, stdout
 
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m%n
 
 
-log4j.category.org.springframework=WARN
-# log4j.category.org.springframework.integration=DEBUG
+log4j.category.org.springframework=ERROR
+#log4j.category.org.springframework.integration.jms=DEBUG
 # log4j.category.org.springframework.integration.jdbc=DEBUG
-log4j.category.org.springframework.jms=DEBUG
+log4j.category.org.springframework.jms=ERROR