Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #584 from garyrussell/INT-2683

Browse Source 
* INT-2683:
  INT-2683, INT-2684 JmsOutboundGateway improvements Added support for caching reply consumers when reply queue is Temporary Queue Added support for reuse of temporary Reply queues in the JmsOutboundGateway
This commit is contained in:
Oleg Zhurakousky
2012-08-17 17:10:59 -04:00
parent 1f59577281 f127dc2d34
commit bfac7999cd
39 changed files with 3330 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java
View File

@@ -47,18 +47,18 @@ import org.springframework.util.Assert;
* Message and sends that Message to a channel. If the 'expectReply' value is
* <code>true</code>, it will also wait for a Spring Integration reply Message
* and convert that into a JMS reply.
*
*
* @author Mark Fisher
* @author Juergen Hoeller
* @author Oleg Zhurakousky
*/
public class ChannelPublishingJmsMessageListener
public class ChannelPublishingJmsMessageListener
implements SessionAwareMessageListener<javax.jms.Message>, InitializingBean, TrackableComponent {
protected final Log logger = LogFactory.getLog(getClass());
private volatile boolean expectReply;
private volatile MessageConverter messageConverter = new SimpleMessageConverter();
private volatile boolean extractRequestPayload = true;
@@ -80,7 +80,7 @@ public class ChannelPublishingJmsMessageListener
private volatile DestinationResolver destinationResolver = new DynamicDestinationResolver();
private volatile JmsHeaderMapper headerMapper = new DefaultJmsHeaderMapper();
private final GatewayDelegate gatewayDelegate = new GatewayDelegate();
/**
@@ -93,27 +93,27 @@ public class ChannelPublishingJmsMessageListener
public void setComponentName(String componentName){
this.gatewayDelegate.setComponentName(componentName);
}
public void setRequestChannel(MessageChannel requestChannel){
this.gatewayDelegate.setRequestChannel(requestChannel);
}
public void setReplyChannel(MessageChannel replyChannel){
this.gatewayDelegate.setReplyChannel(replyChannel);
}
public void setErrorChannel(MessageChannel errorChannel){
this.gatewayDelegate.setErrorChannel(errorChannel);
}
public void setRequestTimeout(long requestTimeout){
this.gatewayDelegate.setRequestTimeout(requestTimeout);
}
public void setReplyTimeout(long replyTimeout){
this.gatewayDelegate.setReplyTimeout(replyTimeout);
}
public void setShouldTrack(boolean shouldTrack) {
this.gatewayDelegate.setShouldTrack(shouldTrack);
}
@@ -125,7 +125,7 @@ public class ChannelPublishingJmsMessageListener
public String getComponentType() {
return this.gatewayDelegate.getComponentType();
}
/**
* Set the default reply destination to send reply messages to. This will
* be applied in case of a request message that does not carry a
@@ -189,7 +189,7 @@ public class ChannelPublishingJmsMessageListener
* JMSMessageID from the request will be copied into the JMSCorrelationID of the reply
* unless there is already a value in the JMSCorrelationID property of the newly created
* reply Message in which case nothing will be copied. If the JMSCorrelationID of the
* request Message should be copied into the JMSCorrelationID of the reply Message
* request Message should be copied into the JMSCorrelationID of the reply Message
* instead, then this value should be set to "JMSCorrelationID".
* Any other value will be treated as a JMS String Property to be copied as-is
* from the request Message into the reply Message with the same property name.
@@ -200,7 +200,7 @@ public class ChannelPublishingJmsMessageListener
/**
* Specify whether explicit QoS should be enabled for replies
* (for timeToLive, priority, and deliveryMode settings).
* (for timeToLive, priority, and deliveryMode settings).
*/
public void setExplicitQosEnabledForReplies(boolean explicitQosEnabledForReplies) {
this.explicitQosEnabledForReplies = explicitQosEnabledForReplies;
@@ -224,7 +224,7 @@ public class ChannelPublishingJmsMessageListener
* converting between JMS Messages and Spring Integration Messages.
* If none is provided, a {@link SimpleMessageConverter} will
* be used.
*
*
* @param messageConverter
*/
public void setMessageConverter(MessageConverter messageConverter) {
@@ -268,10 +268,10 @@ public class ChannelPublishingJmsMessageListener
logger.debug("converted JMS Message [" + jmsMessage + "] to integration Message payload [" + result + "]");
}
}
Map<String, Object> headers = headerMapper.toHeaders(jmsMessage);
Message<?> requestMessage = (result instanceof Message<?>) ?
MessageBuilder.fromMessage((Message<?>) result).copyHeaders(headers).build() :
MessageBuilder.fromMessage((Message<?>) result).copyHeaders(headers).build() :
MessageBuilder.withPayload(result).copyHeaders(headers).build();
if (!this.expectReply) {
this.gatewayDelegate.send(requestMessage);
@@ -292,7 +292,7 @@ public class ChannelPublishingJmsMessageListener
headerMapper.fromHeaders(replyMessage.getHeaders(), jmsReply);
this.copyCorrelationIdFromRequestToReply(jmsMessage, jmsReply);
this.sendReply(jmsReply, destination, session);
}
}
catch (RuntimeException e) {
logger.error("Failed to generate JMS Reply Message from: " + replyResult, e);
throw e;
@@ -308,15 +308,15 @@ public class ChannelPublishingJmsMessageListener
public void afterPropertiesSet() {
this.gatewayDelegate.afterPropertiesSet();
}
protected void start(){
this.gatewayDelegate.start();
}
protected void stop(){
this.gatewayDelegate.stop();
}
private void copyCorrelationIdFromRequestToReply(javax.jms.Message requestMessage, javax.jms.Message replyMessage) throws JMSException {
if (this.correlationKey != null) {
if (this.correlationKey.equals("JMSCorrelationID")) {
@@ -416,17 +416,20 @@ public class ChannelPublishingJmsMessageListener
this.isTopic = isTopic;
}
}
private class GatewayDelegate extends MessagingGatewaySupport {
@Override
protected void send(Object request) {
super.send(request);
}
@Override
protected Message<?> sendAndReceiveMessage(Object request) {
return super.sendAndReceiveMessage(request);
}
@Override
public String getComponentType() {
if (expectReply) {
return "jms:inbound-gateway";

520
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* 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.
@@ -16,6 +16,9 @@
package org.springframework.integration.jms;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -23,6 +26,7 @@ import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
@@ -31,6 +35,7 @@ import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.Topic;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.expression.Expression;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
@@ -40,6 +45,7 @@ import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.connection.ConnectionFactoryUtils;
import org.springframework.jms.support.JmsUtils;
import org.springframework.jms.support.converter.MessageConverter;
@@ -47,6 +53,8 @@ import org.springframework.jms.support.converter.SimpleMessageConverter;
import org.springframework.jms.support.destination.DestinationResolver;
import org.springframework.jms.support.destination.DynamicDestinationResolver;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
/**
* An outbound Messaging Gateway for request/reply JMS.
@@ -58,6 +66,16 @@ import org.springframework.util.Assert;
*/
public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
private final String gatewayId = UUID.randomUUID().toString();
private volatile boolean cachedConsumers;
private volatile Map<Integer, LinkedList<Session>> cachedSessionView;
private volatile boolean replyDestinationExplicitlySet;
private final Map<Long, Destination> tempQueuePerSessionMap = new HashMap<Long, Destination>(10);
private volatile Destination requestDestination;
private volatile String requestDestinationName;
@@ -159,6 +177,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
this.replyPubSubDomain = true;
}
this.replyDestination = replyDestination;
this.replyDestinationExplicitlySet = true;
}
/**
@@ -167,6 +186,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
*/
public void setReplyDestinationName(String replyDestinationName) {
this.replyDestinationName = replyDestinationName;
this.replyDestinationExplicitlySet = true;
}
/**
@@ -240,15 +260,21 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
}
/**
* Provide the name of a JMS property that should hold a generated UUID that
* the receiver of the JMS Message would expect to represent the CorrelationID.
* When waiting for the reply Message, a MessageSelector will be configured
* to match this property name and the UUID value that was sent in the request.
* If this value is NULL (the default) then the reply consumer's MessageSelector
* will be expecting the JMSCorrelationID to equal the Message ID of the request.
* If you want to store the outbound correlation UUID value in the actual
* JMSCorrelationID property, then set this value to "JMSCorrelationID".
* However, any other value will be treated as a JMS String Property.
* Provide the name of a JMS property that should hold a generated JMSCorrelationID.
* If NO value is provided for this attribute, then the reply consumer's
* MessageSelector will be expecting the JMSCorrelationID to equal the Message ID
* of the request. However this would also mean that the MessageSelector will be different
* for each new message, thus requiring a new reply consumer to be created for each Message sent.
* Although this is a very common JMS exchange pattern it is not optimal for high throughput request/reply
* applications. Most MOMs (including Spring Integration's Inbound Gateway) also support propagation
* of the 'JMSCorrelationID' of the consumed request message into the 'JMSCorrelationID' of the newly produced
* reply message. This also allows us to use an optimized value generation algorithm which results in caching
* and reuse of reply consumers resulting in significant performance improvement of the request/reply
* scenarios. So if you have a Spring JMS consumer (e.g., Spring Integration's Inbound Gateway) or
* you know that your MOM supports propagation of the 'CorrelationID' it is highly recommended to set this
* value to 'JMSCorrelationID'. If you set this value to something else the receiving end must now how
* to propagate it. For example: if the receiving end is an inbound-gateway its correlation-key attribute
* must have the same value.
*/
public void setCorrelationKey(String correlationKey) {
this.correlationKey = correlationKey;
@@ -317,48 +343,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
return "jms:outbound-gateway";
}
private Destination getRequestDestination(Message<?> message, Session session) throws JMSException {
if (this.requestDestination != null) {
return this.requestDestination;
}
if (this.requestDestinationName != null) {
return this.resolveRequestDestination(this.requestDestinationName, session);
}
if (this.requestDestinationExpressionProcessor != null) {
Object result = this.requestDestinationExpressionProcessor.processMessage(message);
if (result instanceof Destination) {
return (Destination) result;
}
if (result instanceof String) {
return this.resolveRequestDestination((String) result, session);
}
throw new MessageDeliveryException(message,
"Evaluation of requestDestinationExpression failed to produce a Destination or destination name. Result was: " + result);
}
throw new MessageDeliveryException(message,
"No requestDestination, requestDestinationName, or requestDestinationExpression has been configured.");
}
private Destination resolveRequestDestination(String requestDestinationName, Session session) throws JMSException {
Assert.notNull(this.destinationResolver,
"DestinationResolver is required when relying upon the 'requestDestinationName' property.");
return this.destinationResolver.resolveDestinationName(
session, requestDestinationName, this.requestPubSubDomain);
}
private Destination getReplyDestination(Session session) throws JMSException {
if (this.replyDestination != null) {
return this.replyDestination;
}
if (this.replyDestinationName != null) {
Assert.notNull(this.destinationResolver,
"DestinationResolver is required when relying upon the 'replyDestinationName' property.");
return this.destinationResolver.resolveDestinationName(
session, this.replyDestinationName, this.replyPubSubDomain);
}
return session.createTemporaryQueue();
}
@SuppressWarnings("unchecked")
@Override
public final void onInit() {
synchronized (this.initializationMonitor) {
@@ -376,6 +361,24 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
this.requestDestinationExpressionProcessor.setConversionService(getConversionService());
}
this.initialized = true;
if (this.connectionFactory instanceof CachingConnectionFactory){
this.cachedConsumers = ((CachingConnectionFactory)this.connectionFactory).isCacheConsumers();
DirectFieldAccessor cfDfa = new DirectFieldAccessor(this.connectionFactory);
this.cachedSessionView = (Map<Integer, LinkedList<Session>>) cfDfa.getPropertyValue("cachedSessions");
}
if (this.cachedConsumers && !StringUtils.hasText(this.correlationKey)
&& (this.replyDestination != null || StringUtils.hasText(this.replyDestinationName))){
logger.warn("Caching consumers without custom correlation-key can lead to " +
"significant performance degradation and OutOfMemoryError. Either do not use consumer caching " +
"by setting 'cacheConsumers' attribute to FALSE in the CachingConnectionFactory or set 'correlation-key'" +
"to a value (e.g., correlation-key=\"JMSCorrelationID\")");
}
else if (!this.cachedConsumers && StringUtils.hasText(this.correlationKey)){
logger.warn("Using custom 'correlationKey' without caching consumers can lead to " +
"significant performance degradation. Consider using the CachingConnectionFactory " +
"with 'cacheConsumers' attribute set to TRUE");
}
}
}
@@ -413,12 +416,93 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
}
}
/**
* Create a new JMS Connection for this JMS gateway.
*/
protected Connection createConnection() throws JMSException {
return this.connectionFactory.createConnection();
}
/**
* Create a new JMS Session using the provided Connection.
*/
protected Session createSession(Connection connection) throws JMSException {
return connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
}
private Destination determineRequestDestination(Message<?> message, Session session) throws JMSException {
if (this.requestDestination != null) {
return this.requestDestination;
}
if (this.requestDestinationName != null) {
return this.resolveRequestDestination(this.requestDestinationName, session);
}
if (this.requestDestinationExpressionProcessor != null) {
Object result = this.requestDestinationExpressionProcessor.processMessage(message);
if (result instanceof Destination) {
return (Destination) result;
}
if (result instanceof String) {
return this.resolveRequestDestination((String) result, session);
}
throw new MessageDeliveryException(message,
"Evaluation of requestDestinationExpression failed to produce a Destination or destination name. Result was: " + result);
}
throw new MessageDeliveryException(message,
"No requestDestination, requestDestinationName, or requestDestinationExpression has been configured.");
}
private Destination resolveRequestDestination(String requestDestinationName, Session session) throws JMSException {
Assert.notNull(this.destinationResolver,
"DestinationResolver is required when relying upon the 'requestDestinationName' property.");
return this.destinationResolver.resolveDestinationName(
session, requestDestinationName, this.requestPubSubDomain);
}
private Destination determineReplyDestination(Session session, long sessionId) throws JMSException {
Destination replyDestinationToReturn = this.replyDestination;
if (replyDestinationToReturn == null){
if (StringUtils.hasText(this.replyDestinationName)) {
Assert.notNull(this.destinationResolver,
"DestinationResolver is required when relying upon the 'replyDestinationName' property.");
replyDestinationToReturn = this.destinationResolver.resolveDestinationName(
session, this.replyDestinationName, this.replyPubSubDomain);
}
else if (this.tempQueuePerSessionMap.containsKey(sessionId)){
replyDestinationToReturn = this.tempQueuePerSessionMap.get(sessionId);
}
else if (this.cachedConsumers){
if (StringUtils.hasText(this.correlationKey)){
// will rely on LIKE MessageSelector. Only need one TemporaryQueue for reply
// as if the replyDestinationName was specified
replyDestinationToReturn = session.createTemporaryQueue();
this.replyDestination = replyDestinationToReturn;
}
else {
replyDestinationToReturn = session.createTemporaryQueue();
this.tempQueuePerSessionMap.put(sessionId, replyDestinationToReturn);
}
}
else {
replyDestinationToReturn = session.createTemporaryQueue();
}
}
return replyDestinationToReturn;
}
private javax.jms.Message sendAndReceive(Message<?> requestMessage) throws JMSException {
Connection connection = this.createConnection();
Session session = null;
Destination replyTo = null;
long sessionId = 0;
try {
session = this.createSession(connection);
sessionId = System.identityHashCode(session);
// convert to JMS Message
Object objectToSend = requestMessage;
@@ -431,111 +515,240 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
headerMapper.fromHeaders(requestMessage.getHeaders(), jmsRequest);
// TODO: support a JmsReplyTo header in the SI Message?
replyTo = this.getReplyDestination(session);
jmsRequest.setJMSReplyTo(replyTo);
replyTo = this.determineReplyDestination(session, sessionId);
connection.start();
Integer priority = requestMessage.getHeaders().getPriority();
if (priority == null) {
priority = this.priority;
}
javax.jms.Message replyMessage = null;
Destination requestDestination = this.getRequestDestination(requestMessage, session);
if (this.correlationKey != null) {
replyMessage = this.doSendAndReceiveWithGeneratedCorrelationId(requestDestination, jmsRequest, replyTo, session, priority);
}
else if (replyTo instanceof TemporaryQueue || replyTo instanceof TemporaryTopic) {
replyMessage = this.doSendAndReceiveWithTemporaryReplyToDestination(requestDestination, jmsRequest, replyTo, session, priority);
}
else {
replyMessage = this.doSendAndReceiveWithMessageIdCorrelation(requestDestination, jmsRequest, replyTo, session, priority);
}
Destination requestDestinationToUse = this.determineRequestDestination(requestMessage, session);
javax.jms.Message replyMessage = this.doSendAndReceive(requestDestinationToUse, jmsRequest, replyTo, session, priority, sessionId);
return replyMessage;
}
finally {
JmsUtils.closeSession(session);
this.deleteDestinationIfTemporary(replyTo);
if (!this.replyDestinationExplicitlySet &&
this.isTemporaryDestination(replyTo) &&
!this.isCachedSession(session)) {
this.deleteDestinationIfTemporary(replyTo);
this.tempQueuePerSessionMap.remove(sessionId);
}
ConnectionFactoryUtils.releaseConnection(connection, this.connectionFactory, true);
}
}
/**
* Creates the MessageConsumer before sending the request Message since we are generating our own correlationId value for the MessageSelector.
*/
private javax.jms.Message doSendAndReceiveWithGeneratedCorrelationId(Destination requestDestination,
javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
MessageProducer messageProducer = null;
MessageConsumer messageConsumer = null;
try {
messageProducer = session.createProducer(requestDestination);
String correlationId = UUID.randomUUID().toString().replaceAll("'", "''");
Assert.state(this.correlationKey != null, "correlationKey must not be null");
String messageSelector = null;
if (this.correlationKey.equals("JMSCorrelationID")) {
jmsRequest.setJMSCorrelationID(correlationId);
messageSelector = "JMSCorrelationID = '" + correlationId + "'";
}
else {
jmsRequest.setStringProperty(this.correlationKey, correlationId);
messageSelector = this.correlationKey + " = '" + correlationId + "'";
}
messageConsumer = session.createConsumer(replyTo, messageSelector);
this.sendRequestMessage(jmsRequest, messageProducer, priority);
return this.receiveReplyMessage(messageConsumer);
}
finally {
JmsUtils.closeMessageProducer(messageProducer);
JmsUtils.closeMessageConsumer(messageConsumer);
}
}
/**
* Creates the MessageConsumer before sending the request Message since we do not need any correlation.
*/
private javax.jms.Message doSendAndReceiveWithTemporaryReplyToDestination(Destination requestDestination,
javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
MessageProducer messageProducer = null;
MessageConsumer messageConsumer = null;
try {
messageProducer = session.createProducer(requestDestination);
messageConsumer = session.createConsumer(replyTo);
this.sendRequestMessage(jmsRequest, messageProducer, priority);
return this.receiveReplyMessage(messageConsumer);
}
finally {
JmsUtils.closeMessageProducer(messageProducer);
JmsUtils.closeMessageConsumer(messageConsumer);
}
}
/**
* Creates the MessageConsumer after sending the request Message since we need the MessageID for correlation with a MessageSelector.
*/
private javax.jms.Message doSendAndReceiveWithMessageIdCorrelation(Destination requestDestination,
javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
if (replyTo instanceof Topic && logger.isWarnEnabled()) {
logger.warn("Relying on the MessageID for correlation is not recommended when using a Topic as the replyTo Destination " +
"because that ID can only be provided to a MessageSelector after the reuqest Message has been sent thereby " +
"creating a race condition where a fast response might be sent before the MessageConsumer has been created. " +
"Consider providing a value to the 'correlationKey' property of this gateway instead. Then the MessageConsumer " +
"will be created before the request Message is sent.");
}
MessageProducer messageProducer = null;
MessageConsumer messageConsumer = null;
private javax.jms.Message doSendAndReceive(Destination requestDestinationToUse,
javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority, long sessionId) throws JMSException {
MessageProducer messageProducer = session.createProducer(requestDestinationToUse);
jmsRequest.setJMSReplyTo(replyTo);
try {
messageProducer = session.createProducer(requestDestination);
this.sendRequestMessage(jmsRequest, messageProducer, priority);
String messageId = jmsRequest.getJMSMessageID().replaceAll("'", "''");
String messageSelector = "JMSCorrelationID = '" + messageId + "'";
messageConsumer = session.createConsumer(replyTo, messageSelector);
return this.receiveReplyMessage(messageConsumer);
if (StringUtils.hasText(this.correlationKey)){
return this.exchangeWithCorrelationKey(messageProducer, jmsRequest, session, sessionId, replyTo, priority);
}
else {
return this.exchangeWithoutCorrelationKey(messageProducer, jmsRequest, session, sessionId, replyTo, priority);
}
}
finally {
JmsUtils.closeMessageProducer(messageProducer);
}
}
private javax.jms.Message exchangeWithCorrelationKey(MessageProducer messageProducer, javax.jms.Message jmsRequest,
Session session, long sessionId, Destination replyTo, int priority) throws JMSException {
String messageCorrelationId = ObjectUtils.getIdentityHexString(jmsRequest);
String consumerCorrelationId = gatewayId + "_" + sessionId;
if (this.correlationKey.equals("JMSCorrelationID")) {
jmsRequest.setJMSCorrelationID(consumerCorrelationId + "$" + messageCorrelationId);
}
else {
jmsRequest.setStringProperty(correlationKey, consumerCorrelationId + "$" + messageCorrelationId);
}
String messageSelector = correlationKey + " LIKE '" + consumerCorrelationId + "%'";
MessageConsumer messageConsumer = null;
try {
messageConsumer = this.createMessageConsumer(session, replyTo, messageSelector, sessionId);
this.sendRequestMessage(jmsRequest, messageProducer, priority);
return this.doReceive(messageConsumer, this.correlationKey, messageCorrelationId, sessionId);
}
finally {
JmsUtils.closeMessageConsumer(messageConsumer);
}
}
private javax.jms.Message exchangeWithoutCorrelationKey(MessageProducer messageProducer, javax.jms.Message jmsRequest,
Session session, long sessionId, Destination replyTo, int priority) throws JMSException {
MessageConsumer messageConsumer = null;
try {
String messageCorrelationId = null;
/*
* When using the message id for correlation, we must null any correlation id because we
* may have an upstream gateway waiting on this correlation id; consider:
* ob->ib->ob->ib->ob->ib
* If we propagate the correlation id, the third ob will attempt to correlate on
* the same selector.
*/
if (jmsRequest.getJMSCorrelationID() != null) {
jmsRequest.setJMSCorrelationID(null);
}
if (this.replyDestinationExplicitlySet) {
if (replyTo instanceof Topic && logger.isWarnEnabled()) {
logger.warn("Relying on the MessageID for correlation is not recommended when using a Topic as the replyTo Destination " +
"because that ID can only be provided to a MessageSelector after the request Message has been sent thereby " +
"creating a race condition where a fast response might be sent before the MessageConsumer has been created. " +
"Consider providing a value to the 'correlationKey' property of this gateway instead. Then the MessageConsumer " +
"will be created before the request Message is sent.");
}
this.sendRequestMessage(jmsRequest, messageProducer, priority);
messageCorrelationId = jmsRequest.getJMSMessageID().replaceAll("'", "''");
String messageSelector = "JMSCorrelationID = '" + messageCorrelationId + "'";
messageConsumer = this.createMessageConsumer(session, replyTo, messageSelector, sessionId);
}
else {
messageConsumer = this.createMessageConsumer(session, replyTo, null, sessionId);
this.sendRequestMessage(jmsRequest, messageProducer, priority);
messageCorrelationId = jmsRequest.getJMSMessageID().replaceAll("'", "''");
}
return this.doReceive(messageConsumer, null, messageCorrelationId, sessionId);
}
finally {
JmsUtils.closeMessageConsumer(messageConsumer);
}
}
/**
* Will perform clean up if failures detected during message receive
*/
private javax.jms.Message doReceive(MessageConsumer messageConsumer, String correlationKey, String messageCorrelationIdToMatch,
long sessionId) throws JMSException{
try {
return this.receiveCorrelatedReplyMessage(messageConsumer, correlationKey, messageCorrelationIdToMatch);
}
catch (javax.jms.IllegalStateException e) {
this.clearDestination(sessionId);
throw e;
}
}
private javax.jms.Message receiveCorrelatedReplyMessage(MessageConsumer messageConsumer,
String correlationKey, String messageCorrelationIdToMatch) throws JMSException {
long timeLeft = this.receiveTimeout < 0 ? Long.MAX_VALUE : this.receiveTimeout;
javax.jms.Message replyMessage = null;
boolean replyFound = false;
while (!replyFound && timeLeft > 0){
long startTime = System.currentTimeMillis();
replyMessage = (this.receiveTimeout >= 0) ? messageConsumer.receive(timeLeft) : messageConsumer.receive();
if (replyMessage != null) {
replyFound = checkReplyMessage(correlationKey, messageCorrelationIdToMatch, replyMessage);
if (!replyFound) {
timeLeft -= (System.currentTimeMillis() - startTime);
}
else {
replyMessage.setJMSCorrelationID(null);
}
}
else {
timeLeft = 0;
}
}
return replyFound ? replyMessage : null;
}
/**
* Examine the reply to determine if it is the one we are expecting
* @return true if the reply is correlated.
*/
private boolean checkReplyMessage(String correlationKey, String messageCorrelationIdToMatch,
javax.jms.Message replyMessage) throws JMSException {
String jmsCorrelationId = null;
boolean replyFound;
if (correlationKey != null && !correlationKey.equals("JMSCorrelationID")){
jmsCorrelationId = replyMessage.getStringProperty(correlationKey);
}
else {
jmsCorrelationId = replyMessage.getJMSCorrelationID();
}
if (StringUtils.hasText(jmsCorrelationId) && StringUtils.hasText(messageCorrelationIdToMatch)){
if (jmsCorrelationId.endsWith(messageCorrelationIdToMatch)){
replyFound = true;
}
else {
// Essentially we are discarding the uncorrelated message here and moving on to
// the next one since we can no longer communicate with its originating producer
// since it has already timed out waiting for the reply. We are also honoring the original timeout
if (this.logger.isDebugEnabled()){
this.logger.debug("Discarded late arriving reply: " + replyMessage);
}
replyFound = false;
}
}
else {
replyFound = true;
}
return replyFound;
}
private MessageConsumer createMessageConsumer(Session session, Destination replyTo, String messageSelector, long sessionId) throws JMSException{
MessageConsumer consumer;
try {
if (this.cachedConsumers && this.isTemporaryDestination(replyTo)){
if (StringUtils.hasText(messageSelector)){
consumer = TemporaryConsumerUtils.createConsumer(session, replyTo, messageSelector);
}
else {
consumer = TemporaryConsumerUtils.createConsumer(session, replyTo, null);
}
}
else {
if (StringUtils.hasText(messageSelector)){
consumer = session.createConsumer(replyTo, messageSelector);
}
else {
consumer = session.createConsumer(replyTo);
}
}
}
catch (InvalidDestinationException e) {
this.clearDestination(sessionId);
throw e;
}
return consumer;
}
private void sendRequestMessage(javax.jms.Message jmsRequest, MessageProducer messageProducer, int priority) throws JMSException {
if (this.explicitQosEnabled) {
messageProducer.send(jmsRequest, this.deliveryMode, priority, this.timeToLive);
@@ -545,10 +758,6 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
}
}
private javax.jms.Message receiveReplyMessage(MessageConsumer messageConsumer) throws JMSException {
return (this.receiveTimeout >= 0) ? messageConsumer.receive(receiveTimeout) : messageConsumer.receive();
}
/**
* Deletes either a {@link TemporaryQueue} or {@link TemporaryTopic}.
* Ignores any other {@link Destination} type and also ignores any
@@ -569,17 +778,44 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
}
/**
* Create a new JMS Connection for this JMS gateway.
* Will clear reply destination *only* if such reply destination is temporary
* and was created internally
*/
protected Connection createConnection() throws JMSException {
return this.connectionFactory.createConnection();
private void clearDestination(long sessionId){
if (this.tempQueuePerSessionMap.containsKey(sessionId)){
this.tempQueuePerSessionMap.remove(sessionId);
}
else if (this.isTemporaryDestination(this.replyDestination) && !this.replyDestinationExplicitlySet){
this.replyDestination = null;
}
}
/**
* Create a new JMS Session using the provided Connection.
* Checks if this destination (topic or queue) is a temporary destination
*/
protected Session createSession(Connection connection) throws JMSException {
return connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
private boolean isTemporaryDestination(Destination destination){
return destination instanceof TemporaryQueue || destination instanceof TemporaryTopic;
}
/**
* Validates if this session is cached by the CachingConnectionFactory
* Ignores any exception that may be thrown
*/
private boolean isCachedSession(Session session){
try {
// this.cacheConsumers means we are using CCF
if (this.cachedConsumers && this.cachedSessionView != null &&
this.cachedSessionView.containsKey(Session.AUTO_ACKNOWLEDGE)){
List<Session> cachedSessions = this.cachedSessionView.get(Session.AUTO_ACKNOWLEDGE);
if (cachedSessions != null){
return cachedSessions.contains(session);
}
}
} catch (Exception e) {
if (logger.isTraceEnabled()) {
logger.trace("Exception while detecting cached session", e);
}
}
return false;
}
}

114
spring-integration-jms/src/main/java/org/springframework/integration/jms/TemporaryConsumerUtils.java Normal file
View File

@@ -0,0 +1,114 @@
/*
* 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 java.lang.reflect.Constructor;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Proxy;
import java.util.Map;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.DirectFieldAccessor;
/**
* NOT A PUBLIC API
*
* Temporary class, will be removed as soon as
* https://jira.springsource.org/browse/SPR-9648 is addressed
*
* @author Oleg Zhurakousky
* @since 2.2
*/
class TemporaryConsumerUtils {
private static volatile Constructor<?> consumerCacheKeyConstructor;
private static volatile Constructor<?> cachedMessageConsumerConstructor;
private final static Log logger = LogFactory.getLog(TemporaryConsumerUtils.class);
public static MessageConsumer createConsumer(Session session, Destination destination, String messageSelector)
throws JMSException {
if (Proxy.isProxyClass(session.getClass())){
InvocationHandler invocationHandler = Proxy.getInvocationHandler(session);
DirectFieldAccessor ihDa = new DirectFieldAccessor(invocationHandler);
@SuppressWarnings("unchecked")
Map<Object, MessageConsumer> cachedConsumers =
(Map<Object, MessageConsumer>) ihDa.getPropertyValue("cachedConsumers");
return getCachedConsumer(cachedConsumers, session, destination, messageSelector, false);
}
else {
return session.createConsumer(destination);
}
}
private static MessageConsumer getCachedConsumer(Map<Object, MessageConsumer> cachedConsumers, Session targetSession,
Destination destination, String selector, boolean noLocal) throws JMSException {
Object cacheKey = createConsumerCacheKey(destination, selector, noLocal, null);
MessageConsumer consumer = cachedConsumers.get(cacheKey);
if (consumer != null) {
if (logger.isTraceEnabled()) {
logger.trace("Found cached JMS MessageConsumer for destination [" + destination + "]: " + consumer);
}
}
else {
consumer = targetSession.createConsumer(destination, selector);
if (logger.isDebugEnabled()) {
logger.debug("Creating cached JMS MessageConsumer for destination [" + destination + "]: " + consumer);
}
cachedConsumers.put(cacheKey, consumer);
}
return createCachedMessageConsumer(consumer);
}
private static MessageConsumer createCachedMessageConsumer(MessageConsumer messageConsumer){
try {
if (cachedMessageConsumerConstructor == null){
Class<?> cachedMessageConsumerClass =
Class.forName("org.springframework.jms.connection.CachedMessageConsumer",
true, Thread.currentThread().getContextClassLoader());
cachedMessageConsumerConstructor = cachedMessageConsumerClass.getConstructor(MessageConsumer.class);
cachedMessageConsumerConstructor.setAccessible(true);
}
return (MessageConsumer) cachedMessageConsumerConstructor.newInstance(messageConsumer);
} catch (Exception e) {
throw new IllegalStateException("Failed to create org.springframework.jms.connection.CachedMessageConsumer", e);
}
}
private static Object createConsumerCacheKey(Destination destination, String selector, boolean noLocal, String subscription){
try {
if (consumerCacheKeyConstructor == null){
Class<?> consumerCacheKeyClass =
Class.forName("org.springframework.jms.connection.CachingConnectionFactory$ConsumerCacheKey",
true, Thread.currentThread().getContextClassLoader());
consumerCacheKeyConstructor =
consumerCacheKeyClass.getConstructor(Destination.class, String.class, boolean.class, String.class);
consumerCacheKeyConstructor.setAccessible(true);
}
return consumerCacheKeyConstructor.newInstance(destination, selector, noLocal, subscription);
} catch (Exception e) {
throw new IllegalStateException("Failed to create org.springframework.jms.connection.CachedMessageConsumer", e);
}
}
}

6
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* 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.
@@ -23,6 +23,7 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.util.StringUtils;
/**
@@ -40,8 +41,7 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser {
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.jms.JmsOutboundGateway");
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsOutboundGateway.class);
builder.addPropertyReference("connectionFactory", element.getAttribute("connection-factory"));
String requestDestination = element.getAttribute("request-destination");
String requestDestinationName = element.getAttribute("request-destination-name");

34
spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd
View File

@@ -844,20 +844,29 @@
</xsd:attribute>
<xsd:attribute name="reply-destination-name" type="xsd:string"/>
<xsd:attribute name="reply-pub-sub-domain" type="xsd:string"/>
<xsd:attribute name="correlation-key" type="xsd:string">
<xsd:attribute name="correlation-key">
<xsd:annotation>
<xsd:documentation><![CDATA[
Provide the name of a JMS property that should hold a generated UUID that
the receiver of the JMS Message would expect to represent the CorrelationID.
When waiting for the reply Message, a MessageSelector will be configured
to match this property name and the UUID value that was sent in the request.
Provide the name of a JMS property that should hold a generated CorrelationID.
If NO value is provided for this attribute, then the reply consumer's
MessageSelector will be expecting the JMSCorrelationID to equal the Message ID
of the request. If you want to store the outbound correlation UUID value in the
actual "JMSCorrelationID" property, then set this value to "JMSCorrelationID".
However, any other value will be treated as a JMS String Property.
of the request. However this would also mean that the MessageSelector will be different
for each new message, thus requiring a new reply consumer to be created for each Message sent.
Although this is a very common JMS exchange pattern it is not optimal for high thru-put request/reply
applications. Most MOMs (including Spring Integration's Inbound Gateway) also support propagation
of the 'CorrelationID' of the consumed request message into the 'CorrelationID' of the newly produced
reply message. This also allows us to use an optimized value generation algorithm which results in caching
and reuse of reply consumers resulting in significant performance improvement of the request/reply
scenarios. So if you have a Spring JMS consumer (e.g., Spring Integration's Inbound Gateway) or
you know that your MOM supports propagation of the 'CorrelationID' it is highly recommended to set this
value to 'JMSCorrelationID'. If you set this value to something else the receiving end must now how
to propagate it. For example: if the receiving end is an inbound-gateway its correlation-key attribute
must have the same value.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="correlation-key-enum-type correlation-key-ext"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="destination-resolver" type="xsd:string">
<xsd:annotation>
@@ -958,6 +967,15 @@
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:simpleType name="correlation-key-enum-type">
<xsd:restriction base="xsd:string">
<xsd:enumeration value="JMSCorrelationID"/>
</xsd:restriction>
</xsd:simpleType>
<xsd:simpleType name="correlation-key-ext">
<xsd:restriction base="xsd:string"/>
</xsd:simpleType>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>

5
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsWithMarshallingMessageConverterTests.java
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;
}
}
}

73
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java Normal file
View File

@@ -0,0 +1,73 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.jms.request_reply;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.message.GenericMessage;
import org.springframework.util.StopWatch;
/**
* @author Oleg Zhurakousky
*/
public class MiscellaneousTests {
/**
* Asserts that receive-timeout is honored even if
* requests (once in process), takes less then receive-timeout value
* when requests are queued up (e.g., single consumer receiver)
*/
@Test
public void testTimeoutHonoringWhenRequestsQueuedUp() throws Exception{
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("honor-timeout.xml", this.getClass());
final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
final CountDownLatch latch = new CountDownLatch(3);
final AtomicInteger replies = new AtomicInteger();
StopWatch stopWatch = new StopWatch();
stopWatch.start();
for (int i = 0; i < 3; i++) {
this.exchange(latch, gateway, replies);
}
latch.await();
stopWatch.stop();
assertTrue(stopWatch.getTotalTimeMillis() <= 11000);
assertEquals(1, replies.get());
}
private void exchange(final CountDownLatch latch, final RequestReplyExchanger gateway, final AtomicInteger replies) {
new Thread(new Runnable() {
public void run() {
try {
gateway.exchange(new GenericMessage<String>(""));
replies.incrementAndGet();
} catch (Exception e) {
//ignore
}
latch.countDown();
}
}).start();
}
}

168
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java Normal file
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();
}
}

181
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java Normal file
View File

@@ -0,0 +1,181 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.jms.request_reply;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.message.GenericMessage;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
*/
public class PipelineNamedReplyQueuesJmsTests {
private final Executor executor = Executors.newFixedThreadPool(30);
int requests = 50;
int timeouts;
/**
* jms:out(reply-destination-name="pipeline01-01") -> jms:in -> randomTimeoutProcess ->
* jms:out -> jms:in
*/
@Test
public void testPipeline1() throws Exception{
this.test("pipeline-named-queue-01.xml");
}
/**
* jms:out(reply-destination-name="pipeline02-01") -> jms:in -> randomTimeoutProcess ->
* jms:out(reply-destination-name="pipeline02-02") -> jms:in ->
* jms:out(reply-destination-name="pipeline02-03") -> jms:in
*/
@Test
public void testPipeline2() throws Exception{
this.test("pipeline-named-queue-02.xml");
}
/**
* Same as {@link #testPipeline2()} except all gateways use the same reply queue.
* and zero failures expected (no timeouts on server).
* jms:out(reply-destination-name="pipeline02a-01") -> jms:in -> zeroTimeoutProcess ->
* jms:out(reply-destination-name="pipeline02a-01") -> jms:in ->
* jms:out(reply-destination-name="pipeline02a-01") -> jms:in
*/
@Test
public void testPipeline2a() throws Exception{
this.test("pipeline-named-queue-02a.xml");
assertEquals(0, this.timeouts);
}
/**
* jms:out(reply-destination-name="pipeline03-01", correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess ->
* jms:out(reply-destination-name="pipeline03-02") -> jms:in ->
* jms:out(reply-destination-name="pipeline03-03") -> jms:in
*/
@Test
public void testPipeline3() throws Exception{
this.test("pipeline-named-queue-03.xml");
}
/**
* Same as {@link #testPipeline3()} except all gateways use the same reply queue.
* Ensures the correlation id is not propagated. No timeouts expected.
* jms:out(reply-destination-name="pipeline03a-01", correlation-key="JMSCorrelationID") -> jms:in -> zeroTimeoutProcess ->
* jms:out(reply-destination-name="pipeline03a-01") -> jms:in ->
* jms:out(reply-destination-name="pipeline03a-01") -> jms:in
* Ensures reply came from service after third gateway
*/
@Test
public void testPipeline3a() throws Exception{
this.test("pipeline-named-queue-03a.xml", 20000);
assertEquals(0, this.timeouts);
}
/**
* jms:out(reply-destination-name="pipeline04-01", correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess ->
* jms:out(reply-destination-name="pipeline04-02") -> jms:in
*/
@Test
public void testPipeline4() throws Exception{
this.test("pipeline-named-queue-04.xml");
}
/**
* jms:out(reply-destination-name="pipeline05-01", correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess ->
* jms:out(reply-destination-name="pipeline05-02", correlation-key="JMSCorrelationID") -> jms:in
*/
@Test
public void testPipeline5() throws Exception{
this.test("pipeline-named-queue-05.xml");
}
/**
* jms:out(reply-destination-name="pipeline06-01", correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess ->
* jms:out(reply-destination-name="pipeline06-02", correlation-key="foo") -> jms:in(correlation-key="foo")
*/
@Test
public void testPipeline6() throws Exception{
this.test("pipeline-named-queue-06.xml");
}
/**
* jms:out(reply-destination-name="pipeline07-01", correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess ->
* jms:out(reply-destination-name="pipeline07-02", correlation-key="foo") -> jms:in(correlation-key="foo")
*/
@Test
public void testPipeline7() throws Exception{
this.test("pipeline-named-queue-07.xml");
}
public void test(String contextConfig) throws Exception {
test(contextConfig, 0);
}
public void test(String contextConfig, final int offset) throws Exception {
this.timeouts = 0;
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(contextConfig, this.getClass());
final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class);
final CountDownLatch latch = new CountDownLatch(requests);
final AtomicInteger successCounter = new AtomicInteger();
final AtomicInteger timeoutCounter = new AtomicInteger();
final AtomicInteger failureCounter = new AtomicInteger();
for (int i = 0; i < requests; i++) {
final int y = i;
executor.execute(new Runnable() {
public void run() {
try {
assertEquals(y + offset, gateway.exchange(new GenericMessage<Integer>(y)).getPayload());
successCounter.incrementAndGet();
} catch (MessageTimeoutException e) {
timeoutCounter.incrementAndGet();
} catch (Throwable t) {
t.printStackTrace();
failureCounter.incrementAndGet();
} finally {
latch.countDown();
}
}
});
}
assertTrue(latch.await(60, TimeUnit.SECONDS));
System.out.println("Success: " + successCounter.get());
System.out.println("Timeout: " + timeoutCounter.get());
System.out.println("Failure: " + failureCounter.get());
// technically all we care that its > 0,
// but reality of this test it has to be something more then 0
assertTrue(successCounter.get() > 10);
assertEquals(0, failureCounter.get());
assertEquals(requests, successCounter.get() + timeoutCounter.get());
this.timeouts = timeoutCounter.get();
context.destroy();
}
}

246
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCachedConsumersTests.java Normal file
View File

@@ -0,0 +1,246 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.jms.request_reply;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
import java.util.concurrent.CountDownLatch;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.junit.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.SessionAwareMessageListener;
import org.springframework.jms.support.converter.SimpleMessageConverter;
/**
* @author Oleg Zhurakousky
*/
public class RequestReplyScenariosWithCachedConsumersTests {
private final SimpleMessageConverter converter = new SimpleMessageConverter();
@Test(expected=MessageTimeoutException.class)
public void messageCorrelationBasedOnRequestMessageIdOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("standardMessageIdCopyingConsumerWithOptimization", RequestReplyExchanger.class);
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
final Destination requestDestination = context.getBean("siOutQueueOptimizedA", Destination.class);
final Destination replyDestination = context.getBean("siInQueueOptimizedA", Destination.class);
new Thread(new Runnable() {
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
});
}
}).start();
gateway.exchange(new GenericMessage<String>("foo"));
context.close();
}
@Test
public void messageCorrelationBasedOnRequestMessageIdNonOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("standardMessageIdCopyingConsumerWithoutOptimization", RequestReplyExchanger.class);
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
final Destination requestDestination = context.getBean("siOutQueueNonOptimizedB", Destination.class);
final Destination replyDestination = context.getBean("siInQueueNonOptimizedB", Destination.class);
new Thread(new Runnable() {
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
});
}
}).start();
org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
context.close();
}
@Test
public void messageCorrelationBasedOnRequestCorrelationIdOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("correlationPropagatingConsumerWithOptimization", RequestReplyExchanger.class);
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
final Destination requestDestination = context.getBean("siOutQueueOptimizedC", Destination.class);
final Destination replyDestination = context.getBean("siInQueueOptimizedC", Destination.class);
new Thread(new Runnable() {
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
}
});
}
}).start();
org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
context.close();
}
@Test(expected=MessageTimeoutException.class)
public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("correlationPropagatingConsumerWithoutOptimization", RequestReplyExchanger.class);
CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
final Destination requestDestination = context.getBean("siOutQueueNonOptimizedD", Destination.class);
final Destination replyDestination = context.getBean("siInQueueNonOptimizedD", Destination.class);
new Thread(new Runnable() {
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
}
});
}
}).start();
org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
context.close();
}
@Test
public void messageCorrelationBasedOnRequestCorrelationIdTimedOutFirstReplyOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ClassPathXmlApplicationContext context =
new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass());
RequestReplyExchanger gateway =
context.getBean("correlationPropagatingConsumerWithOptimizationDelayFirstReply", RequestReplyExchanger.class);
final ConnectionFactory connectionFactory = context.getBean("connectionFactory", ConnectionFactory.class);
final Destination requestDestination = context.getBean("siOutQueueE", Destination.class);
final Destination replyDestination = context.getBean("siInQueueE", Destination.class);
for (int i = 0; i < 3; i++) {
System.out.println("#### " + i);
try {
gateway.exchange(gateway.exchange(new GenericMessage<String>("foo")));
} catch (Exception e) {/*ignore*/}
}
final CountDownLatch latch = new CountDownLatch(1);
new Thread(new Runnable() {
public void run() {
DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
dmlc.setConnectionFactory(connectionFactory);
dmlc.setDestination(requestDestination);
dmlc.setMessageListener(new SessionAwareMessageListener<Message>() {
public void onMessage(Message message, Session session) {
String requestPayload = (String) extractPayload(message);
try {
TextMessage replyMessage = session.createTextMessage();
replyMessage.setText(requestPayload);
replyMessage.setJMSCorrelationID(message.getJMSCorrelationID());
MessageProducer producer = session.createProducer(replyDestination);
producer.send(replyMessage);
} catch (Exception e) {
// ignore. the test will fail
}
}
});
dmlc.afterPropertiesSet();
dmlc.start();
latch.countDown();
}
}).start();
latch.await();
TestUtils.getPropertyValue(context.getBean("fastGateway"), "handler", JmsOutboundGateway.class).setReceiveTimeout(10000);
Thread.sleep(1000);
assertEquals("bar", gateway.exchange(new GenericMessage<String>("bar")).getPayload());
context.close();
}
private Object extractPayload(Message jmsMessage) {
try {
return converter.fromMessage(jmsMessage);
} catch (Exception e) {
e.printStackTrace();
fail();
}
return null;
}
}

84
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java Normal file
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;
}
}
}

157
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithNonCachedConsumersTests.java Normal file
View File

@@ -0,0 +1,157 @@
/*
* Copyright 2002-2012 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.jms.request_reply;
import static org.junit.Assert.assertEquals;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.integration.jms.config.ActiveMqTestUtils;
import org.springframework.integration.message.GenericMessage;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
/**
* @author Oleg Zhurakousky
*/
public class RequestReplyScenariosWithNonCachedConsumersTests {
@Test(expected=MessageTimeoutException.class)
public void messageCorrelationBasedOnRequestMessageIdOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("optimizedMessageId", RequestReplyExchanger.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
final Destination requestDestination = context.getBean("siOutQueueC", Destination.class);
final Destination replyDestination = context.getBean("siInQueueC", Destination.class);
new Thread(new Runnable() {
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
});
}
}).start();
org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
}
@Test
public void messageCorrelationBasedOnRequestMessageIdNonOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("nonoptimizedMessageId", RequestReplyExchanger.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
final Destination requestDestination = context.getBean("siOutQueueD", Destination.class);
final Destination replyDestination = context.getBean("siInQueueD", Destination.class);
new Thread(new Runnable() {
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSMessageID());
return message;
}
});
}
}).start();
org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
}
@Test
public void messageCorrelationBasedOnRequestCorrelationIdOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("optimized", RequestReplyExchanger.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
final Destination requestDestination = context.getBean("siOutQueueA", Destination.class);
final Destination replyDestination = context.getBean("siInQueueA", Destination.class);
new Thread(new Runnable() {
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
}
});
}
}).start();
org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
}
@Test(expected=MessageTimeoutException.class)
public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() throws Exception{
ActiveMqTestUtils.prepare();
ApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass());
RequestReplyExchanger gateway = context.getBean("nonoptimized", RequestReplyExchanger.class);
ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
final Destination requestDestination = context.getBean("siOutQueueB", Destination.class);
final Destination replyDestination = context.getBean("siInQueueB", Destination.class);
new Thread(new Runnable() {
public void run() {
final Message requestMessage = jmsTemplate.receive(requestDestination);
jmsTemplate.send(replyDestination, new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage();
message.setText("bar");
message.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
return message;
}
});
}
}).start();
org.springframework.integration.Message<?> siReplyMessage = gateway.exchange(new GenericMessage<String>("foo"));
assertEquals("bar", siReplyMessage.getPayload());
}
}

275
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java Normal file
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;
}
}

44
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/broken-broker.xml Normal file
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>

82
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml Normal file
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>

44
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/honor-timeout.xml Normal file
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>

49
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/mult-producer-and-consumers-temp-reply.xml Normal file
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>

58
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-01.xml Normal file
View File

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

59
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-02.xml Normal file
View File

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

60
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-03.xml Normal file
View File

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

59
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-04.xml Normal file
View File

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

60
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-05.xml Normal file
View File

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

60
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-06.xml Normal file
View File

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

61
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-07.xml Normal file
View File

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

61
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-08.xml Normal file
View File

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

62
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-09.xml Normal file
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="pipeline09" default-request-timeout="10000" default-reply-timeout="10000"/>
<int-jms:outbound-gateway request-channel="pipeline09"
connection-factory="connectionFactory"
request-destination-name="pipeline09-queue-01"
correlation-key="foo"/>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="pipeline09-queue-01"
connection-factory="connectionFactory"
concurrent-consumers="10"
reply-timeout="10000"
correlation-key="foo"/>
<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
<int:header-enricher>
<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
</int:header-enricher>
<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
<int:transformer expression="payload"/>
</int:chain>
<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
connection-factory="connectionFactory"
request-destination-name="pipeline09-queue-02"
correlation-key="bar"/>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="pipeline09-queue-02"
connection-factory="connectionFactory"
concurrent-consumers="10"
reply-timeout="10000"
correlation-key="bar"/>
<int:transformer input-channel="anotherIn" expression="payload"/>
<bean id="connectionFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

59
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-01.xml Normal file
View File

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

73
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02.xml Normal file
View File

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

73
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02a.xml Normal file
View File

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

74
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03.xml Normal file
View File

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

77
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml Normal file
View File

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

62
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml Normal file
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="pipeline04" default-request-timeout="10000" default-reply-timeout="10000"/>
<int-jms:outbound-gateway request-channel="pipeline04"
connection-factory="connectionFactory"
reply-destination-name="pipeline04-01"
correlation-key="foo"
request-destination-name="siOutQueue"/>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue"
connection-factory="connectionFactory"
concurrent-consumers="10"
correlation-key="foo"
reply-timeout="10000"/>
<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
<int:header-enricher>
<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
</int:header-enricher>
<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
<int:transformer expression="payload"/>
</int:chain>
<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
connection-factory="connectionFactory"
reply-destination-name="pipeline04-02"
request-destination-name="anotherGatewayQueue"/>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue"
connection-factory="connectionFactory"
concurrent-consumers="10"
reply-timeout="10000"/>
<int:transformer input-channel="anotherIn" expression="payload"/>
<bean id="connectionFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

63
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-05.xml Normal file
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="pipeline05" default-request-timeout="10000" default-reply-timeout="10000"/>
<int-jms:outbound-gateway request-channel="pipeline05"
connection-factory="connectionFactory"
reply-destination-name="pipeline05-01"
correlation-key="foo"
request-destination-name="siOutQueue"/>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue"
connection-factory="connectionFactory"
concurrent-consumers="10"
correlation-key="foo"
reply-timeout="10000"/>
<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
<int:header-enricher>
<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
</int:header-enricher>
<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
<int:transformer expression="payload"/>
</int:chain>
<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
connection-factory="connectionFactory"
reply-destination-name="pipeline05-02"
request-destination-name="anotherGatewayQueue"
correlation-key="JMSCorrelationID"/>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue"
connection-factory="connectionFactory"
concurrent-consumers="10"
reply-timeout="10000"/>
<int:transformer input-channel="anotherIn" expression="payload"/>
<bean id="connectionFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

64
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-06.xml Normal file
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"
reply-destination-name="pipeline06-01"
correlation-key="JMSCorrelationID"
request-destination-name="siOutQueue"/>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue"
connection-factory="connectionFactory"
concurrent-consumers="10"
correlation-key="JMSCorrelationID"
reply-timeout="10000"/>
<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
<int:header-enricher>
<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
</int:header-enricher>
<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
<int:transformer expression="payload"/>
</int:chain>
<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
connection-factory="connectionFactory"
reply-destination-name="pipeline06-02"
request-destination-name="anotherGatewayQueue"
correlation-key="foo"/>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue"
connection-factory="connectionFactory"
concurrent-consumers="10"
reply-timeout="10000"
correlation-key="foo"/>
<int:transformer input-channel="anotherIn" expression="payload"/>
<bean id="connectionFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

64
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-07.xml Normal file
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="pipeline07" default-request-timeout="10000" default-reply-timeout="10000"/>
<int-jms:outbound-gateway request-channel="pipeline07"
connection-factory="connectionFactory"
reply-destination-name="pipeline07-01"
correlation-key="foo"
request-destination-name="siOutQueue"/>
<int-jms:inbound-gateway request-channel="jmsIn"
request-destination-name="siOutQueue"
connection-factory="connectionFactory"
concurrent-consumers="10"
correlation-key="foo"
reply-timeout="10000"/>
<int:chain input-channel="jmsIn" output-channel="anotherGatewayChannel">
<int:header-enricher>
<int:header name="delay" expression="new java.util.Random().nextInt(3000)"/>
</int:header-enricher>
<int:delayer id="foo" default-delay="0" delay-header-name="delay"/>
<int:transformer expression="payload"/>
</int:chain>
<int-jms:outbound-gateway request-channel="anotherGatewayChannel"
connection-factory="connectionFactory"
reply-destination-name="pipeline07-02"
request-destination-name="anotherGatewayQueue"
correlation-key="bar"/>
<int-jms:inbound-gateway request-channel="anotherIn"
request-destination-name="anotherGatewayQueue"
connection-factory="connectionFactory"
concurrent-consumers="10"
reply-timeout="10000"
correlation-key="bar"/>
<int:transformer input-channel="anotherIn" expression="payload"/>
<bean id="connectionFactory"
class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="true" />
</bean>
</beans>

107
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-cached-consumers.xml Normal file
View File

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

87
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-no-cached-consumers.xml Normal file
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"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.2.xsd">
<int:gateway id="optimized" default-request-channel="jmsInOptimized"/>
<int-jms:outbound-gateway request-channel="jmsInOptimized"
connection-factory="connectionFactory"
request-destination="siOutQueueA"
reply-destination="siInQueueA"
correlation-key="JMSCorrelationID"/>
<bean id="siOutQueueA" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="siOutQueueA.not.cached"/>
</bean>
<bean id="siInQueueA" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="siInQueueA.not.cached"/>
</bean>
<!-- -->
<int:gateway id="nonoptimized" default-request-channel="jmsInNonOptimized"/>
<int-jms:outbound-gateway request-channel="jmsInNonOptimized"
connection-factory="connectionFactory"
request-destination="siOutQueueB"
reply-destination="siInQueueB"/>
<bean id="siOutQueueB" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="siOutQueueB.not.cached"/>
</bean>
<bean id="siInQueueB" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="siInQueueB.not.cached"/>
</bean>
<!-- -->
<int:gateway id="optimizedMessageId" default-request-channel="jmsInOptimizedC"/>
<int-jms:outbound-gateway request-channel="jmsInOptimizedC"
connection-factory="connectionFactory"
request-destination="siOutQueueC"
reply-destination="siInQueueC"
correlation-key="JMSCorrelationID"/>
<bean id="siOutQueueC" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="siOutQueueC.not.cached"/>
</bean>
<bean id="siInQueueC" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="siInQueueC.not.cached"/>
</bean>
<!-- -->
<int:gateway id="nonoptimizedMessageId" default-request-channel="jmsInNonOptimizedD"/>
<int-jms:outbound-gateway request-channel="jmsInNonOptimizedD"
connection-factory="connectionFactory"
request-destination="siOutQueueD"
reply-destination="siInQueueD"/>
<bean id="siOutQueueD" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="siOutQueueD.not.cached"/>
</bean>
<bean id="siInQueueD" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="siInQueueD.not.cached"/>
</bean>
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost"/>
</bean>
</property>
<property name="cacheProducers" value="true" />
<property name="cacheConsumers" value="false" />
<property name="sessionCacheSize" value="10" />
</bean>
</beans>

30
spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-temp-reply-consumers.xml Normal file
View File

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

8
spring-integration-jms/src/test/resources/log4j.properties
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