diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index e939582866..d88b55f555 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -47,18 +47,18 @@ import org.springframework.util.Assert; * Message and sends that Message to a channel. If the 'expectReply' value is * true, 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, 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 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,20 +416,17 @@ 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"; diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java index 50e93f207e..91de46dbe3 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2011 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,9 +16,6 @@ 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; @@ -26,7 +23,6 @@ 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; @@ -35,7 +31,6 @@ 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; @@ -45,7 +40,6 @@ 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; @@ -53,8 +47,6 @@ 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. @@ -66,16 +58,6 @@ import org.springframework.util.StringUtils; */ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { - private final String gatewayId = UUID.randomUUID().toString(); - - private volatile boolean cachedConsumers; - - private volatile Map> cachedSessionView; - - private volatile boolean replyDestinationExplicitlySet; - - private final Map tempQueuePerSessionMap = new HashMap(10); - private volatile Destination requestDestination; private volatile String requestDestinationName; @@ -177,7 +159,6 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { this.replyPubSubDomain = true; } this.replyDestination = replyDestination; - this.replyDestinationExplicitlySet = true; } /** @@ -186,7 +167,6 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { */ public void setReplyDestinationName(String replyDestinationName) { this.replyDestinationName = replyDestinationName; - this.replyDestinationExplicitlySet = true; } /** @@ -260,21 +240,15 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { } /** - * 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. + * 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. */ public void setCorrelationKey(String correlationKey) { this.correlationKey = correlationKey; @@ -343,7 +317,48 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { return "jms:outbound-gateway"; } - @SuppressWarnings("unchecked") + 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(); + } + @Override public final void onInit() { synchronized (this.initializationMonitor) { @@ -361,24 +376,6 @@ 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>) 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"); - } } } @@ -416,93 +413,12 @@ 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; @@ -515,240 +431,111 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { headerMapper.fromHeaders(requestMessage.getHeaders(), jmsRequest); // TODO: support a JmsReplyTo header in the SI Message? - replyTo = this.determineReplyDestination(session, sessionId); - + replyTo = this.getReplyDestination(session); + jmsRequest.setJMSReplyTo(replyTo); connection.start(); Integer priority = requestMessage.getHeaders().getPriority(); if (priority == null) { priority = this.priority; } - - Destination requestDestinationToUse = this.determineRequestDestination(requestMessage, session); - - javax.jms.Message replyMessage = this.doSendAndReceive(requestDestinationToUse, jmsRequest, replyTo, session, priority, sessionId); - + 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); + } return replyMessage; } finally { JmsUtils.closeSession(session); - - if (!this.replyDestinationExplicitlySet && - this.isTemporaryDestination(replyTo) && - !this.isCachedSession(session)) { - - this.deleteDestinationIfTemporary(replyTo); - this.tempQueuePerSessionMap.remove(sessionId); - } - + this.deleteDestinationIfTemporary(replyTo); 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 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); + 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; try { - if (StringUtils.hasText(this.correlationKey)){ - return this.exchangeWithCorrelationKey(messageProducer, jmsRequest, session, sessionId, replyTo, priority); - } - else { - return this.exchangeWithoutCorrelationKey(messageProducer, jmsRequest, session, sessionId, replyTo, priority); - } + 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); } 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); @@ -758,6 +545,10 @@ 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 @@ -778,44 +569,17 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { } /** - * Will clear reply destination *only* if such reply destination is temporary - * and was created internally + * Create a new JMS Connection for this JMS gateway. */ - 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; - } + protected Connection createConnection() throws JMSException { + return this.connectionFactory.createConnection(); } /** - * Checks if this destination (topic or queue) is a temporary destination + * Create a new JMS Session using the provided Connection. */ - private boolean isTemporaryDestination(Destination destination){ - return destination instanceof TemporaryQueue || destination instanceof TemporaryTopic; + protected Session createSession(Connection connection) throws JMSException { + return connection.createSession(false, Session.AUTO_ACKNOWLEDGE); } - /** - * 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 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; - } } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/TemporaryConsumerUtils.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/TemporaryConsumerUtils.java deleted file mode 100644 index 9f3ff846a2..0000000000 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/TemporaryConsumerUtils.java +++ /dev/null @@ -1,114 +0,0 @@ -/* - * 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 cachedConsumers = - (Map) ihDa.getPropertyValue("cachedConsumers"); - return getCachedConsumer(cachedConsumers, session, destination, messageSelector, false); - } - else { - return session.createConsumer(destination); - } - } - - private static MessageConsumer getCachedConsumer(Map 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); - } - } -} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java index d689f52d2b..77ffbb1b60 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2011 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,7 +23,6 @@ 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; /** @@ -41,7 +40,8 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser { @Override protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsOutboundGateway.class); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( + "org.springframework.integration.jms.JmsOutboundGateway"); builder.addPropertyReference("connectionFactory", element.getAttribute("connection-factory")); String requestDestination = element.getAttribute("request-destination"); String requestDestinationName = element.getAttribute("request-destination-name"); diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd index 640707a8c3..5c53c83dfe 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.2.xsd @@ -844,29 +844,20 @@ - + - - - @@ -967,15 +958,6 @@ - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsWithMarshallingMessageConverterTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsWithMarshallingMessageConverterTests.java index c2d3bd411d..7682e03a10 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsWithMarshallingMessageConverterTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsWithMarshallingMessageConverterTests.java @@ -44,7 +44,7 @@ import org.springframework.oxm.XmlMappingException; * @author Oleg Zhurakousky */ public class JmsWithMarshallingMessageConverterTests { - + @Test @SuppressWarnings("unchecked") public void demoWithMarshallingConverter() { @@ -58,6 +58,7 @@ 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()); } @@ -85,7 +86,7 @@ public class JmsWithMarshallingMessageConverterTests { public boolean supports(Class clazz) { return true; - } + } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java deleted file mode 100644 index 64384b98c2..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java +++ /dev/null @@ -1,73 +0,0 @@ -/* - * 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("")); - replies.incrementAndGet(); - } catch (Exception e) { - //ignore - } - latch.countDown(); - } - }).start(); - } -} diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java deleted file mode 100644 index 922b0375f5..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java +++ /dev/null @@ -1,168 +0,0 @@ -/* - * 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(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(); - } -} diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java deleted file mode 100644 index 2438b9989f..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java +++ /dev/null @@ -1,181 +0,0 @@ -/* - * 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(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(); - } -} diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCachedConsumersTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCachedConsumersTests.java deleted file mode 100644 index 2d6678e512..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCachedConsumersTests.java +++ /dev/null @@ -1,246 +0,0 @@ -/* - * 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("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("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("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("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("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() { - - 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("bar")).getPayload()); - - context.close(); - } - - private Object extractPayload(Message jmsMessage) { - try { - return converter.fromMessage(jmsMessage); - } catch (Exception e) { - e.printStackTrace(); - fail(); - } - return null; - } -} diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java deleted file mode 100644 index 1a9c34ae06..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java +++ /dev/null @@ -1,84 +0,0 @@ -/* - * 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; - } - } -} diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithNonCachedConsumersTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithNonCachedConsumersTests.java deleted file mode 100644 index 2c276e9d70..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithNonCachedConsumersTests.java +++ /dev/null @@ -1,157 +0,0 @@ -/* - * 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("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("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("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("foo")); - assertEquals("bar", siReplyMessage.getPayload()); - } -} diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java deleted file mode 100644 index 78909846d0..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java +++ /dev/null @@ -1,275 +0,0 @@ -/* - * 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("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() { - - 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("foo")); - } catch (Exception e) { - // ignore - } - Thread.sleep(1000); - try { - assertEquals("bar", gateway.exchange(new GenericMessage("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.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.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 list = new ArrayList(); - 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; - } -} diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/broken-broker.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/broken-broker.xml deleted file mode 100644 index 219613d57a..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/broken-broker.xml +++ /dev/null @@ -1,44 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml deleted file mode 100644 index 23f68c040e..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml +++ /dev/null @@ -1,82 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/honor-timeout.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/honor-timeout.xml deleted file mode 100644 index 00337c6b17..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/honor-timeout.xml +++ /dev/null @@ -1,44 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/mult-producer-and-consumers-temp-reply.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/mult-producer-and-consumers-temp-reply.xml deleted file mode 100644 index a74d179060..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/mult-producer-and-consumers-temp-reply.xml +++ /dev/null @@ -1,49 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-01.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-01.xml deleted file mode 100644 index 9b7e7a6c18..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-01.xml +++ /dev/null @@ -1,58 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-02.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-02.xml deleted file mode 100644 index 65d7b7c7bb..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-02.xml +++ /dev/null @@ -1,59 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-03.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-03.xml deleted file mode 100644 index 88bc5726a5..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-03.xml +++ /dev/null @@ -1,60 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-04.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-04.xml deleted file mode 100644 index 9b53d64b28..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-04.xml +++ /dev/null @@ -1,59 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-05.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-05.xml deleted file mode 100644 index d02001ebb1..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-05.xml +++ /dev/null @@ -1,60 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-06.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-06.xml deleted file mode 100644 index 9cd95a2327..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-06.xml +++ /dev/null @@ -1,60 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-07.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-07.xml deleted file mode 100644 index 3405e72a8e..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-07.xml +++ /dev/null @@ -1,61 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-08.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-08.xml deleted file mode 100644 index be52c7228a..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-08.xml +++ /dev/null @@ -1,61 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-09.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-09.xml deleted file mode 100644 index 27de5c069f..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-09.xml +++ /dev/null @@ -1,62 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-01.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-01.xml deleted file mode 100644 index 3b22f9dc54..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-01.xml +++ /dev/null @@ -1,59 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02.xml deleted file mode 100644 index 6d24912b93..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02.xml +++ /dev/null @@ -1,73 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02a.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02a.xml deleted file mode 100644 index c5d57c02a0..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02a.xml +++ /dev/null @@ -1,73 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03.xml deleted file mode 100644 index 26aa375f0e..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03.xml +++ /dev/null @@ -1,74 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml deleted file mode 100644 index d003e5b599..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml +++ /dev/null @@ -1,77 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml deleted file mode 100644 index c3e2c43647..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml +++ /dev/null @@ -1,62 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-05.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-05.xml deleted file mode 100644 index 1c97010234..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-05.xml +++ /dev/null @@ -1,63 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-06.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-06.xml deleted file mode 100644 index e97a510c94..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-06.xml +++ /dev/null @@ -1,64 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-07.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-07.xml deleted file mode 100644 index 46db195c75..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-07.xml +++ /dev/null @@ -1,64 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-cached-consumers.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-cached-consumers.xml deleted file mode 100644 index 7aeb3c62f8..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-cached-consumers.xml +++ /dev/null @@ -1,107 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-no-cached-consumers.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-no-cached-consumers.xml deleted file mode 100644 index 29a9a30d38..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-no-cached-consumers.xml +++ /dev/null @@ -1,87 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-temp-reply-consumers.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-temp-reply-consumers.xml deleted file mode 100644 index 565af787e9..0000000000 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-temp-reply-consumers.xml +++ /dev/null @@ -1,30 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - diff --git a/spring-integration-jms/src/test/resources/log4j.properties b/spring-integration-jms/src/test/resources/log4j.properties index b8ebc43ebe..0d11d824ba 100644 --- a/spring-integration-jms/src/test/resources/log4j.properties +++ b/spring-integration-jms/src/test/resources/log4j.properties @@ -1,11 +1,11 @@ -log4j.rootCategory=ERROR, stdout +log4j.rootCategory=WARN, 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=ERROR -#log4j.category.org.springframework.integration.jms=DEBUG +log4j.category.org.springframework=WARN +# log4j.category.org.springframework.integration=DEBUG # log4j.category.org.springframework.integration.jdbc=DEBUG -log4j.category.org.springframework.jms=ERROR +log4j.category.org.springframework.jms=DEBUG