INT-2683 Add Reply Listener Container Option

INT-2683 Tests

Expanded JMS Gateway Tests

INT-2683 First commit

Listener Container Option for Replies

INT-2683 Remove no correlation-key Option

INT-2863 Support DestName and Temp Dest

INT-2683 Add back Support No CorrelationKey

INT-2683 Fix Tests

INT-2683 Polishing

Fallback if no correlationKey and fixed reply queue.

INT-2683 Polishing

Change from SMLC to DMLC

INT-2683 Polishing

Since we changed to DMLC, we can now remove the check for
SingleConnectionFactory.

INT-2683 <reply-listener/> Namespace Support

Add <reply-listener/> subelement to JMS Outbound Gateway.
This commit is contained in:
Gary Russell
2012-08-21 13:47:59 -04:00
committed by Oleg Zhurakousky
parent 9e8d1c83f8
commit 49a6079d1a
45 changed files with 4216 additions and 35 deletions

View File

@@ -0,0 +1,149 @@
/*
* 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;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQQueue;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.message.GenericMessage;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* @author Gary Russell
* @since 2.2
*
*/
public class OutboundGatewayConnectionTests {
private Destination requestQueue1 = new ActiveMQQueue("request1");
private Destination replyQueue1 = new ActiveMQQueue("reply1");
@Test @Ignore // need a more reliable stop/start for AMQ
public void testContainerWithDestBrokenConnection() throws Exception {
BeanFactory beanFactory = mock(BeanFactory.class);
when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.initialize();
when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
BrokerService broker = new BrokerService();
broker.addConnector("tcp://localhost:61616?broker.persistent=false");
broker.start();
ActiveMQConnectionFactory amqConnectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
ConnectionFactory connectionFactory = new CachingConnectionFactory(amqConnectionFactory);
gateway.setConnectionFactory(connectionFactory);
gateway.setRequestDestination(requestQueue1);
gateway.setReplyDestination(replyQueue1);
gateway.setCorrelationKey("JMSCorrelationID");
gateway.setUseReplyContainer(true);
gateway.afterPropertiesSet();
gateway.start();
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(amqConnectionFactory);
template.setReceiveTimeout(5000);
javax.jms.Message request = template.receive(requestQueue1);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
});
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
broker.stop();
broker = new BrokerService();
broker.addConnector("tcp://localhost:61616?broker.persistent=false");
broker.start();
final CountDownLatch latch3 = new CountDownLatch(1);
final CountDownLatch latch4 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
latch3.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch4.countDown();
}
}
});
assertTrue(latch3.await(10, TimeUnit.SECONDS));
template = new JmsTemplate();
template.setConnectionFactory(amqConnectionFactory);
template.setReceiveTimeout(5000);
request = template.receive(requestQueue1);
assertNotNull(request);
final javax.jms.Message jmsReply2 = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return jmsReply2;
}
});
assertTrue(latch4.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
broker.stop();
gateway.stop();
}
}

View File

@@ -0,0 +1,385 @@
/*
* 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;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.message.GenericMessage;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
/**
* @author Gary Russell
* @since 2.2
*
*/
public class OutboundGatewayFunctionTests {
private static Destination requestQueue1 = new ActiveMQQueue("request1");
private static Destination replyQueue1 = new ActiveMQQueue("reply1");
private static Destination requestQueue2 = new ActiveMQQueue("request2");
private static Destination replyQueue2 = new ActiveMQQueue("reply2");
private static Destination requestQueue3 = new ActiveMQQueue("request3");
private static Destination replyQueue3 = new ActiveMQQueue("reply3");
private static Destination requestQueue4 = new ActiveMQQueue("request4");
private static Destination replyQueue4 = new ActiveMQQueue("reply4");
private static Destination requestQueue5 = new ActiveMQQueue("request5");
private static Destination replyQueue5 = new ActiveMQQueue("reply5");
private static Destination requestQueue6 = new ActiveMQQueue("request6");
private static Destination replyQueue6 = new ActiveMQQueue("reply6");
@Test
public void testContainerWithDest() throws Exception {
BeanFactory beanFactory = mock(BeanFactory.class);
when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.initialize();
when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getGatewayConnectionFactory());
gateway.setRequestDestination(requestQueue1);
gateway.setReplyDestination(replyQueue1);
gateway.setCorrelationKey("JMSCorrelationID");
gateway.setUseReplyContainer(true);
gateway.afterPropertiesSet();
gateway.start();
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getTemplateConnectionFactory());
template.setReceiveTimeout(5000);
javax.jms.Message request = template.receive(requestQueue1);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
});
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
gateway.stop();
}
@Test
public void testContainerWithDestNoCorrelation() throws Exception {
BeanFactory beanFactory = mock(BeanFactory.class);
when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.initialize();
when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getGatewayConnectionFactory());
gateway.setRequestDestination(requestQueue2);
gateway.setReplyDestination(replyQueue2);
gateway.setUseReplyContainer(true);
gateway.afterPropertiesSet();
gateway.start();
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getTemplateConnectionFactory());
template.setReceiveTimeout(5000);
javax.jms.Message request = template.receive(requestQueue2);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
}
});
assertTrue(latch2.await(20, TimeUnit.SECONDS));
assertNotNull(reply.get());
gateway.stop();
}
@Test
public void testContainerWithDestName() throws Exception {
BeanFactory beanFactory = mock(BeanFactory.class);
when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.initialize();
when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getGatewayConnectionFactory());
gateway.setRequestDestination(requestQueue3);
gateway.setReplyDestinationName("reply3");
gateway.setCorrelationKey("JMSCorrelationID");
gateway.setUseReplyContainer(true);
gateway.afterPropertiesSet();
gateway.start();
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getTemplateConnectionFactory());
template.setReceiveTimeout(5000);
javax.jms.Message request = template.receive(requestQueue3);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
});
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
gateway.stop();
}
@Test
public void testContainerWithDestNameNoCorrelation() throws Exception {
BeanFactory beanFactory = mock(BeanFactory.class);
when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.initialize();
when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getGatewayConnectionFactory());
gateway.setRequestDestination(requestQueue4);
gateway.setReplyDestinationName("reply4");
gateway.setUseReplyContainer(true);
gateway.afterPropertiesSet();
gateway.start();
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getTemplateConnectionFactory());
template.setReceiveTimeout(5000);
javax.jms.Message request = template.receive(requestQueue4);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
}
});
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
gateway.stop();
}
@Test
public void testContainerWithTemporary() throws Exception {
BeanFactory beanFactory = mock(BeanFactory.class);
when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.initialize();
when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getGatewayConnectionFactory());
gateway.setRequestDestination(requestQueue5);
gateway.setCorrelationKey("JMSCorrelationID");
gateway.setUseReplyContainer(true);
gateway.afterPropertiesSet();
gateway.start();
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getTemplateConnectionFactory());
template.setReceiveTimeout(5000);
javax.jms.Message request = template.receive(requestQueue5);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return jmsReply;
}
});
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
gateway.stop();
}
@Test
public void testContainerWithTemporaryNoCorrelation() throws Exception {
BeanFactory beanFactory = mock(BeanFactory.class);
when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true);
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.initialize();
when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class))
.thenReturn(scheduler);
final JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setBeanFactory(beanFactory);
gateway.setConnectionFactory(getGatewayConnectionFactory());
gateway.setRequestDestination(requestQueue6);
gateway.setUseReplyContainer(true);
gateway.afterPropertiesSet();
gateway.start();
final AtomicReference<Object> reply = new AtomicReference<Object>();
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
latch1.countDown();
try {
reply.set(gateway.handleRequestMessage(new GenericMessage<String>("foo")));
}
finally {
latch2.countDown();
}
}
});
assertTrue(latch1.await(10, TimeUnit.SECONDS));
JmsTemplate template = new JmsTemplate();
template.setConnectionFactory(getTemplateConnectionFactory());
template.setReceiveTimeout(5000);
javax.jms.Message request = template.receive(requestQueue6);
assertNotNull(request);
final javax.jms.Message jmsReply = request;
template.send(request.getJMSReplyTo(), new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID());
return jmsReply;
}
});
assertTrue(latch2.await(10, TimeUnit.SECONDS));
assertNotNull(reply.get());
gateway.stop();
}
private ConnectionFactory getTemplateConnectionFactory() {
ConnectionFactory amqConnectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
return amqConnectionFactory;
}
private ConnectionFactory getGatewayConnectionFactory() {
ConnectionFactory amqConnectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
return new CachingConnectionFactory(amqConnectionFactory);
}
}

View File

@@ -46,6 +46,7 @@ import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.integration.jms.StubMessageConverter;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.support.converter.MessageConverter;
/**
@@ -68,6 +69,17 @@ public class JmsOutboundGatewayParserTests {
accessor = new DirectFieldAccessor(gateway);
int deliveryMode = (Integer)accessor.getPropertyValue("deliveryMode");
assertEquals(DeliveryMode.PERSISTENT, deliveryMode);
DefaultMessageListenerContainer container = TestUtils.getPropertyValue(gateway, "replyContainer",
DefaultMessageListenerContainer.class);
assertEquals(4, TestUtils.getPropertyValue(container, "concurrentConsumers"));
assertEquals(5, TestUtils.getPropertyValue(container, "maxConcurrentConsumers"));
assertEquals(10, TestUtils.getPropertyValue(container, "maxMessagesPerTask"));
assertEquals(2000L, TestUtils.getPropertyValue(container, "receiveTimeout"));
assertEquals(10000L, TestUtils.getPropertyValue(container, "recoveryInterval"));
assertEquals(7, TestUtils.getPropertyValue(container, "idleConsumerLimit"));
assertEquals(2, TestUtils.getPropertyValue(container, "idleTaskExecutionLimit"));
assertEquals(3, TestUtils.getPropertyValue(container, "cacheLevel"));
assertTrue(container.isSessionTransacted());
}
@Test
@@ -78,6 +90,7 @@ public class JmsOutboundGatewayParserTests {
JmsOutboundGateway gateway = TestUtils.getPropertyValue(endpoint, "handler", JmsOutboundGateway.class);
gateway.handleMessage(new GenericMessage<String>("foo"));
assertEquals(1, adviceCalled);
assertEquals(3, TestUtils.getPropertyValue(gateway, "replyContainer.sessionAcknowledgeMode"));
}
@Test

View File

@@ -28,7 +28,6 @@ import javax.jms.TextMessage;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
@@ -111,6 +110,7 @@ public class JmsPriorityTests {
String text = "priority=" + request.getJMSPriority();
TextMessage reply = session.createTextMessage(text);
MessageProducer producer = session.createProducer(request.getJMSReplyTo());
reply.setJMSCorrelationID(request.getJMSMessageID());
producer.send(reply);
}
}

View File

@@ -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;
}
}
}

View File

@@ -15,8 +15,19 @@
<jms:outbound-gateway id="jmsGateway"
request-destination-name="requestQueue"
request-channel="requestChannel"
delivery-persistent="true"/>
delivery-persistent="true">
<jms:reply-listener
acknowledge="transacted"
concurrent-consumers="4"
max-concurrent-consumers="5"
max-messages-per-task="10"
receive-timeout="2000"
recovery-interval="10000"
idle-consumer-limit="7"
idle-task-execution-limit="2"
cache-level="3" />
</jms:outbound-gateway>
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<constructor-arg>
<bean class="org.springframework.integration.jms.StubConnection">
@@ -32,6 +43,16 @@
<jms:request-handler-advice-chain>
<bean class="org.springframework.integration.jms.config.JmsOutboundGatewayParserTests$FooAdvice" />
</jms:request-handler-advice-chain>
<jms:reply-listener
acknowledge="dups-ok"
concurrent-consumers="3"
max-concurrent-consumers="5"
max-messages-per-task="10"
receive-timeout="2000"
recovery-interval="10000"
idle-consumer-limit="1"
idle-task-execution-limit="2"
cache-level="3" />
</jms:outbound-gateway>
</beans>

View File

@@ -0,0 +1,72 @@
/*
* 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() <= 12000);
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();
}
}

View File

@@ -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();
}
}

View File

@@ -0,0 +1,186 @@
/*
* 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 AtomicInteger successCounter = new AtomicInteger();
final AtomicInteger timeoutCounter = new AtomicInteger();
final AtomicInteger failureCounter = new AtomicInteger();
try {
final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
final CountDownLatch latch = new CountDownLatch(requests);
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));
// 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();
}
finally {
System.out.println(contextConfig);
System.out.println("Success: " + successCounter.get());
System.out.println("Timeout: " + timeoutCounter.get());
System.out.println("Failure: " + failureCounter.get());
context.destroy();
}
}
}

View File

@@ -0,0 +1,265 @@
/*
* 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());
try {
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"));
}
finally {
context.destroy();
}
}
@Test
public void messageCorrelationBasedOnRequestMessageIdNonOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
try {
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());
}
finally {
context.destroy();
}
}
@Test
public void messageCorrelationBasedOnRequestCorrelationIdOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
try {
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());
}
finally {
context.destroy();
}
}
@Test(expected=MessageTimeoutException.class)
public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
try {
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());
}
finally {
context.destroy();
}
}
@Test
public void messageCorrelationBasedOnRequestCorrelationIdTimedOutFirstReplyOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
try {
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());
}
finally {
context.destroy();
}
}
private Object extractPayload(Message jmsMessage) {
try {
return converter.fromMessage(jmsMessage);
} catch (Exception e) {
e.printStackTrace();
fail();
}
return null;
}
}

View File

@@ -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;
}
}
}

View File

@@ -0,0 +1,175 @@
/*
* 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.support.AbstractApplicationContext;
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();
AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
try {
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());
}
finally {
context.destroy();
}
}
@Test
public void messageCorrelationBasedOnRequestMessageIdNonOptimized() throws Exception{
ActiveMqTestUtils.prepare();
AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
try {
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());
}
finally {
context.destroy();
}
}
@Test
public void messageCorrelationBasedOnRequestCorrelationIdOptimized() throws Exception{
ActiveMqTestUtils.prepare();
AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
try {
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());
}
finally {
context.destroy();
}
}
@Test(expected=MessageTimeoutException.class)
public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() throws Exception{
ActiveMqTestUtils.prepare();
AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
try {
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());
}
finally {
context.destroy();
}
}
}

View File

@@ -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;
}
}

View File

@@ -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>

View File

@@ -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>

View File

@@ -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>

View File

@@ -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>

View File

@@ -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="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:reply-listener />
</int-jms:outbound-gateway>
<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:reply-listener />
</int-jms:outbound-gateway>
<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>

View File

@@ -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="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:reply-listener />
</int-jms:outbound-gateway>
<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:reply-listener />
</int-jms:outbound-gateway>
<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>

View File

@@ -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="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:reply-listener />
</int-jms:outbound-gateway>
<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:reply-listener />
</int-jms:outbound-gateway>
<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>

View File

@@ -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="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:reply-listener />
</int-jms:outbound-gateway>
<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:reply-listener />
</int-jms:outbound-gateway>
<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>

View File

@@ -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="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:reply-listener />
</int-jms:outbound-gateway>
<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:reply-listener />
</int-jms:outbound-gateway>
<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>

View File

@@ -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"
request-destination-name="pipeline06-queue-01">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<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:reply-listener />
</int-jms:outbound-gateway>
<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>

View File

@@ -0,0 +1,65 @@
<?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:reply-listener />
</int-jms:outbound-gateway>
<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:reply-listener />
</int-jms:outbound-gateway>
<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>

View File

@@ -0,0 +1,65 @@
<?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:reply-listener />
</int-jms:outbound-gateway>
<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:reply-listener />
</int-jms:outbound-gateway>
<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>

View File

@@ -0,0 +1,66 @@
<?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:reply-listener />
</int-jms:outbound-gateway>
<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:reply-listener />
</int-jms:outbound-gateway>
<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>

View File

@@ -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="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="siOutQueue01-01">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue01-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="anotherGatewayQueue01-01">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue01-01"
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://localhosti?broker.persistent=false" />
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

View File

@@ -0,0 +1,83 @@
<?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"
correlation-key="corr02"
request-destination-name="siOutQueue02">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue02"
correlation-key="corr02"
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"
correlation-key="corr02"
request-destination-name="anotherGatewayQueue02">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue02"
correlation-key="corr02"
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="thirdGatewayQueue02">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="thirdIn"
request-destination-name="thirdGatewayQueue02"
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://localhosti?broker.persistent=false" />
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

View File

@@ -0,0 +1,85 @@
<?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"
correlation-key="corr02a"
request-destination-name="siOutQueue02a">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue02a"
correlation-key="corr02a"
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"
correlation-key="corr02a"
request-destination-name="anotherGatewayQueue02a">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue02a"
correlation-key="corr02a"
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"
correlation-key="corr02a"
request-destination-name="thirdGatewayQueue02a">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="thirdIn"
request-destination-name="thirdGatewayQueue02a"
correlation-key="corr02a"
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://localhosti?broker.persistent=false" />
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

View File

@@ -0,0 +1,84 @@
<?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="siOutQueue03">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue03"
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"
correlation-key="corr03"
request-destination-name="anotherGatewayQueue03">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue03"
correlation-key="corr03"
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"
correlation-key="corr03"
request-destination-name="thirdGatewayQueue03">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="thirdIn"
request-destination-name="thirdGatewayQueue03"
correlation-key="corr03"
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?broker.persistent=false" />
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

View File

@@ -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"
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="siOutQueue03a">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue03a"
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"
correlation-key="corr03a"
request-destination-name="anotherGatewayQueue03a">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue03a"
correlation-key="corr03a"
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"
correlation-key="corr03a"
connection-factory="connectionFactory"
reply-destination-name="pipeline03a-01"
request-destination-name="thirdGatewayQueue03a">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int:transformer input-channel="add10kOnTheWayBack" expression="payload + 10000" />
<int-jms:inbound-gateway request-channel="thirdIn"
request-destination-name="thirdGatewayQueue03a"
correlation-key="corr03a"
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?broker.persistent=false" />
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

View File

@@ -0,0 +1,66 @@
<?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="siOutQueue04">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue04"
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="anotherGatewayQueue04">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue04"
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?broker.persistent=false" />
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

View File

@@ -0,0 +1,67 @@
<?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="siOutQueue05">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue05"
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="anotherGatewayQueue05"
correlation-key="JMSCorrelationID">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue05"
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?broker.persistent=false" />
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

View File

@@ -0,0 +1,68 @@
<?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="siOutQueue06">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue06"
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="anotherGatewayQueue06"
correlation-key="foo">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue06"
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?broker.persistent=false" />
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

View File

@@ -0,0 +1,68 @@
<?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="siOutQueue07">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue07"
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="anotherGatewayQueue07"
correlation-key="bar">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue07"
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?broker.persistent=false" />
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

View File

@@ -0,0 +1,118 @@
<?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">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<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">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<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">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<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">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<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">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<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?broker.persistent=false" />
</bean>
</property>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
<property name="sessionCacheSize" value="10" />
</bean>
</beans>

View File

@@ -0,0 +1,95 @@
<?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">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<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">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<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">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<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">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<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>

View File

@@ -0,0 +1,32 @@
<?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">
<int-jms:reply-listener />
</int-jms:outbound-gateway>
<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>

View File

@@ -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