From 49a6079d1adc906d53643fb10488f898faabb9d7 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 21 Aug 2012 13:47:59 -0400 Subject: [PATCH] INT-2683 Add Reply Listener Container Option INT-2683 Tests Expanded JMS Gateway Tests INT-2683 First commit Listener Container Option for Replies INT-2683 Remove no correlation-key Option INT-2863 Support DestName and Temp Dest INT-2683 Add back Support No CorrelationKey INT-2683 Fix Tests INT-2683 Polishing Fallback if no correlationKey and fixed reply queue. INT-2683 Polishing Change from SMLC to DMLC INT-2683 Polishing Since we changed to DMLC, we can now remove the check for SingleConnectionFactory. INT-2683 Namespace Support Add subelement to JMS Outbound Gateway. --- build.gradle | 3 +- .../endpoint/EventDrivenConsumer.java | 18 +- .../integration/endpoint/PollingConsumer.java | 29 +- .../integration/jms/JmsOutboundGateway.java | 608 +++++++++++++++++- .../jms/config/JmsOutboundGatewayParser.java | 38 +- .../jms/config/spring-integration-jms-2.2.xsd | 32 +- .../jms/OutboundGatewayConnectionTests.java | 149 +++++ .../jms/OutboundGatewayFunctionTests.java | 385 +++++++++++ .../config/JmsOutboundGatewayParserTests.java | 13 + .../jms/config/JmsPriorityTests.java | 2 +- ...sWithMarshallingMessageConverterTests.java | 5 +- ...sOutboundGatewayWithDeliveryPersistent.xml | 25 +- .../jms/request_reply/MiscellaneousTests.java | 72 +++ .../jms/request_reply/PipelineJmsTests.java | 168 +++++ .../PipelineNamedReplyQueuesJmsTests.java | 186 ++++++ ...eplyScenariosWithCachedConsumersTests.java | 265 ++++++++ ...nariosWithCorrelationKeyProvidedTests.java | 84 +++ ...yScenariosWithNonCachedConsumersTests.java | 175 +++++ ...eplyScenariosWithTempReplyQueuesTests.java | 275 ++++++++ .../jms/request_reply/broken-broker.xml | 44 ++ .../explicit-correlation-key.xml | 82 +++ .../jms/request_reply/honor-timeout.xml | 44 ++ ...mult-producer-and-consumers-temp-reply.xml | 49 ++ .../jms/request_reply/pipeline-01.xml | 62 ++ .../jms/request_reply/pipeline-02.xml | 63 ++ .../jms/request_reply/pipeline-03.xml | 64 ++ .../jms/request_reply/pipeline-04.xml | 63 ++ .../jms/request_reply/pipeline-05.xml | 64 ++ .../jms/request_reply/pipeline-06.xml | 64 ++ .../jms/request_reply/pipeline-07.xml | 65 ++ .../jms/request_reply/pipeline-08.xml | 65 ++ .../jms/request_reply/pipeline-09.xml | 66 ++ .../request_reply/pipeline-named-queue-01.xml | 63 ++ .../request_reply/pipeline-named-queue-02.xml | 83 +++ .../pipeline-named-queue-02a.xml | 85 +++ .../request_reply/pipeline-named-queue-03.xml | 84 +++ .../pipeline-named-queue-03a.xml | 87 +++ .../request_reply/pipeline-named-queue-04.xml | 66 ++ .../request_reply/pipeline-named-queue-05.xml | 67 ++ .../request_reply/pipeline-named-queue-06.xml | 68 ++ .../request_reply/pipeline-named-queue-07.xml | 68 ++ .../producer-cached-consumers.xml | 118 ++++ .../producer-no-cached-consumers.xml | 95 +++ .../producer-temp-reply-consumers.xml | 32 + .../src/test/resources/log4j.properties | 8 +- 45 files changed, 4216 insertions(+), 35 deletions(-) create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCachedConsumersTests.java create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithNonCachedConsumersTests.java create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/broken-broker.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/honor-timeout.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/mult-producer-and-consumers-temp-reply.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-01.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-02.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-03.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-04.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-05.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-06.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-07.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-08.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-09.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-01.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02a.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-05.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-06.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-07.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-cached-consumers.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-no-cached-consumers.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-temp-reply-consumers.xml diff --git a/build.gradle b/build.gradle index 3e9a7817e9..c7fe00428b 100644 --- a/build.gradle +++ b/build.gradle @@ -483,8 +483,9 @@ project('spring-integration-jms') { 'org.springframework.context;version="[3.1.1, 4.0.0)"', 'org.springframework.core.*;version="[3.1.1, 4.0.0)"', 'org.springframework.jms.*;version="[3.1.1, 4.0.0)"', + 'org.springframework.scheduling.*;version="[3.1.1, 4.0.0)"', 'org.springframework.transaction.*;version="[3.1.1, 4.0.0)"', - 'org.springframework.util;version="[3.1.1, 4.0.0)"', + 'org.springframework.util.*;version="[3.1.1, 4.0.0)"', 'org.apache.commons.logging;version="[1.1.1, 2.0.0)"', 'javax.jms;version="[1.1.0, 2.0.0)";resolution:=optional', 'org.w3c.dom.*;version="0"' diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java index 67164bccbd..1ffc5c9053 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,6 +16,7 @@ package org.springframework.integration.endpoint; +import org.springframework.context.Lifecycle; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.context.NamedComponent; import org.springframework.integration.core.MessageHandler; @@ -25,9 +26,10 @@ import org.springframework.util.StringUtils; /** * Message Endpoint that connects any {@link MessageHandler} implementation to a {@link SubscribableChannel}. - * + * * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ public class EventDrivenConsumer extends AbstractEndpoint { @@ -45,18 +47,24 @@ public class EventDrivenConsumer extends AbstractEndpoint { } - @Override + @Override protected void doStart() { this.logComponentSubscriptionEvent(true); this.inputChannel.subscribe(this.handler); + if (this.handler instanceof Lifecycle) { + ((Lifecycle) this.handler).start(); + } } - @Override + @Override protected void doStop() { this.logComponentSubscriptionEvent(false); this.inputChannel.unsubscribe(this.handler); + if (this.handler instanceof Lifecycle) { + ((Lifecycle) this.handler).stop(); + } } - + private void logComponentSubscriptionEvent(boolean add){ if (this.handler instanceof NamedComponent && this.inputChannel instanceof NamedComponent){ String channelName = ((NamedComponent)this.inputChannel).getComponentName(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index 427bb6ea7b..d7d555544e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,7 +18,7 @@ package org.springframework.integration.endpoint; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - +import org.springframework.context.Lifecycle; import org.springframework.integration.Message; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.PollableChannel; @@ -27,20 +27,21 @@ import org.springframework.util.Assert; /** * Message Endpoint that connects any {@link MessageHandler} implementation * to a {@link PollableChannel}. - * + * * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ public class PollingConsumer extends AbstractPollingEndpoint { private final Log logger = LogFactory.getLog(this.getClass()); - + private final PollableChannel inputChannel; private final MessageHandler handler; private volatile long receiveTimeout = 1000; - + public PollingConsumer(PollableChannel inputChannel, MessageHandler handler) { Assert.notNull(inputChannel, "inputChannel must not be null"); Assert.notNull(handler, "handler must not be null"); @@ -53,6 +54,24 @@ public class PollingConsumer extends AbstractPollingEndpoint { this.receiveTimeout = receiveTimeout; } + @Override + protected void doStart() { + if (this.handler instanceof Lifecycle) { + ((Lifecycle) this.handler).start(); + } + super.doStart(); + } + + + @Override + protected void doStop() { + if (this.handler instanceof Lifecycle) { + ((Lifecycle) this.handler).stop(); + } + super.doStop(); + } + + @Override protected boolean doPoll() { Message message = (this.receiveTimeout >= 0) 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 91de46dbe3..409b742079 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-2011 the original author or authors. + * Copyright 2002-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,8 +16,17 @@ package org.springframework.integration.jms; +import java.util.Date; +import java.util.HashMap; +import java.util.Iterator; import java.util.Map; +import java.util.Map.Entry; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import javax.jms.Connection; import javax.jms.ConnectionFactory; @@ -25,12 +34,14 @@ import javax.jms.DeliveryMode; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageConsumer; +import javax.jms.MessageListener; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TemporaryQueue; import javax.jms.TemporaryTopic; import javax.jms.Topic; +import org.springframework.context.SmartLifecycle; import org.springframework.expression.Expression; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; @@ -41,12 +52,14 @@ import org.springframework.integration.handler.AbstractReplyProducingMessageHand import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor; import org.springframework.integration.support.MessageBuilder; import org.springframework.jms.connection.ConnectionFactoryUtils; +import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.jms.support.JmsUtils; import org.springframework.jms.support.converter.MessageConverter; 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.StringUtils; /** * An outbound Messaging Gateway for request/reply JMS. @@ -55,8 +68,9 @@ import org.springframework.util.Assert; * @author Arjen Poutsma * @author Juergen Hoeller * @author Oleg Zhurakousky + * @author Gary Russell */ -public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { +public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler implements SmartLifecycle, MessageListener { private volatile Destination requestDestination; @@ -98,8 +112,31 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { private volatile boolean initialized; + private volatile GatewayReplyListenerContainer replyContainer; + + private volatile ReplyContainerProperties replyContainerProperties; + + private volatile boolean useReplyContainer; + private final Object initializationMonitor = new Object(); + private volatile boolean autoStartup; + + private volatile boolean active; + + private final AtomicLong correlationId = new AtomicLong(); + + private final String gatewayCorrelation = UUID.randomUUID().toString(); + + private final Map> replies = + new HashMap>(); + + private final ConcurrentHashMap earlyOrLateReplies = + new ConcurrentHashMap(); + + private volatile ScheduledFuture reaper; + + private final Object lifeCycleMonitor = new Object(); /** * Set whether message delivery should be persistent or non-persistent, @@ -312,11 +349,26 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { this.setOutputChannel(replyChannel); } + /** + * @param replyContainerProperties the replyContainerproperties to set + */ + public void setReplyContainerProperties(ReplyContainerProperties replyContainerProperties) { + this.replyContainerProperties = replyContainerProperties; + this.useReplyContainer = true; + } + @Override public String getComponentType() { return "jms:outbound-gateway"; } + /** + * @param useReplyContainer the useReplyContainer to set + */ + public void setUseReplyContainer(boolean useReplyContainer) { + this.useReplyContainer = useReplyContainer; + } + private Destination getRequestDestination(Message message, Session session) throws JMSException { if (this.requestDestination != null) { return this.requestDestination; @@ -346,7 +398,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { session, requestDestinationName, this.requestPubSubDomain); } - private Destination getReplyDestination(Session session) throws JMSException { + private Destination determineReplyDestination(Session session) throws JMSException { if (this.replyDestination != null) { return this.replyDestination; } @@ -359,6 +411,18 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { return session.createTemporaryQueue(); } + public int getPhase() { + return Integer.MAX_VALUE; + } + + public boolean isAutoStartup() { + return this.autoStartup; + } + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + @Override public final void onInit() { synchronized (this.initializationMonitor) { @@ -375,10 +439,119 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { this.requestDestinationExpressionProcessor.setBeanFactory(getBeanFactory()); this.requestDestinationExpressionProcessor.setConversionService(getConversionService()); } + /* + * This is needed because there is no way to detect 2 or more gateways using the same reply queue + * with no correlation key. + */ + if (this.useReplyContainer && (this.correlationKey == null && + (this.replyDestination != null || this.replyDestinationName != null))) { + if (logger.isWarnEnabled()) { + logger.warn("The gateway cannot use a reply listener container with a specified destination(Name) " + + "without a 'correlation-key'; " + + "a container will NOT be used; " + + "to avoid this problem, set the 'correlation-key' attribute; " + + "some consumers, including the Spring Integration , " + + "support the use of the value 'JMSCorrelationID' " + + "for this purpose. Alternatively, do not specify a reply destination " + + "and a temporary queue will be used for replies."); + } + this.useReplyContainer = false; + } + if (this.useReplyContainer) { + GatewayReplyListenerContainer container = new GatewayReplyListenerContainer(); + setContainerProperties(container); + container.afterPropertiesSet(); + this.replyContainer = container; + } this.initialized = true; } } + private void setContainerProperties(GatewayReplyListenerContainer container) { + container.setConnectionFactory(this.connectionFactory); + if (this.replyDestination != null) { + container.setDestination(this.replyDestination); + } + if (StringUtils.hasText(this.replyDestinationName)) { + container.setDestinationName(this.replyDestinationName); + } + if (this.destinationResolver != null) { + container.setDestinationResolver(this.destinationResolver); + } + container.setPubSubDomain(this.replyPubSubDomain); + if (this.correlationKey != null) { + String messageSelector = this.correlationKey + " LIKE '" + this.gatewayCorrelation + "%'"; + container.setMessageSelector(messageSelector); + } + container.setMessageListener(this); + if (this.replyContainerProperties != null) { + if (this.replyContainerProperties.isSessionTransacted() != null) { + container.setSessionTransacted(this.replyContainerProperties.isSessionTransacted()); + } + if (this.replyContainerProperties.getCacheLevel() != null) { + container.setCacheLevel(this.replyContainerProperties.getCacheLevel()); + } + if (this.replyContainerProperties.getConcurrentConsumers() != null) { + container.setConcurrentConsumers(this.replyContainerProperties.getConcurrentConsumers()); + } + if (this.replyContainerProperties.getIdleConsumerLimit() != null) { + container.setIdleConsumerLimit(this.replyContainerProperties.getIdleConsumerLimit()); + } + if (this.replyContainerProperties.getIdleTaskExecutionLimit() != null) { + container.setIdleTaskExecutionLimit(this.replyContainerProperties.getIdleTaskExecutionLimit()); + } + if (this.replyContainerProperties.getMaxConcurrentConsumers() != null) { + container.setMaxConcurrentConsumers(this.replyContainerProperties.getMaxConcurrentConsumers()); + } + if (this.replyContainerProperties.getMaxMessagesPerTask() != null) { + container.setMaxMessagesPerTask(this.replyContainerProperties.getMaxMessagesPerTask()); + } + if (this.replyContainerProperties.getReceiveTimeout() != null) { + container.setReceiveTimeout(this.replyContainerProperties.getReceiveTimeout()); + } + if (this.replyContainerProperties.getRecoveryInterval() != null) { + container.setRecoveryInterval(this.replyContainerProperties.getRecoveryInterval()); + } + if (this.replyContainerProperties.getSessionAcknowledgeMode() != null) { + container.setSessionAcknowledgeMode(this.replyContainerProperties.getSessionAcknowledgeMode()); + } + } + } + + public void start() { + synchronized (this.lifeCycleMonitor) { + if (!this.active) { + if (this.replyContainer != null) { + this.replyContainer.start(); + if (this.receiveTimeout >= 0) { + this.reaper = this.getTaskScheduler().schedule(new LateReplyReaper(), new Date()); + } + } + this.active = true; + } + } + } + + public void stop() { + synchronized (this.lifeCycleMonitor) { + if (this.replyContainer != null) { + this.replyContainer.stop(); + this.deleteDestinationIfTemporary(this.replyContainer.getDestination()); + this.reaper.cancel(false); + } + this.active = false; + } + } + + public boolean isRunning() { + return this.active; + } + + public void stop(Runnable callback) { + this.stop(); + callback.run(); + } + @Override protected Object handleRequestMessage(final Message message) { if (!this.initialized) { @@ -386,7 +559,13 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { } final Message requestMessage = MessageBuilder.fromMessage(message).build(); try { - javax.jms.Message jmsReply = this.sendAndReceive(requestMessage); + javax.jms.Message jmsReply; + if (this.replyContainer == null) { + jmsReply = this.sendAndReceiveWithoutContainer(requestMessage); + } + else { + jmsReply = this.sendAndReceiveWithContainer(requestMessage); + } if (jmsReply == null) { throw new MessageTimeoutException(message, "failed to receive JMS response within timeout of: " + this.receiveTimeout + "ms"); @@ -399,6 +578,11 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { } } Map jmsReplyHeaders = this.headerMapper.toHeaders(jmsReply); + + if (this.replyContainer != null && this.correlationKey != null) { + // do not propagate back the gateway's internal correlation id + jmsReplyHeaders.remove(this.correlationKey); + } Message replyMessage = null; if (result instanceof Message){ replyMessage = MessageBuilder.fromMessage((Message) result).copyHeaders(jmsReplyHeaders).build(); @@ -413,7 +597,60 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { } } - private javax.jms.Message sendAndReceive(Message requestMessage) throws JMSException { + private javax.jms.Message sendAndReceiveWithContainer(Message requestMessage) throws JMSException { + Connection connection = this.createConnection(); + Session session = null; + Destination replyTo = this.replyContainer.getReplyDestination(); + try { + session = this.createSession(connection); + + // convert to JMS Message + Object objectToSend = requestMessage; + if (this.extractRequestPayload) { + objectToSend = requestMessage.getPayload(); + } + javax.jms.Message jmsRequest = this.messageConverter.toMessage(objectToSend, session); + + // map headers + headerMapper.fromHeaders(requestMessage.getHeaders(), jmsRequest); + + jmsRequest.setJMSReplyTo(replyTo); + connection.start(); + + Integer priority = requestMessage.getHeaders().getPriority(); + if (priority == null) { + priority = this.priority; + } + Destination requestDestination = this.getRequestDestination(requestMessage, session); + + /* + * Remove any existing correlation id that was mapped from the inbound message + * (it will be restored by normal ARPMH header processing). + */ + jmsRequest.setJMSCorrelationID(null); + javax.jms.Message reply = null; + if (this.correlationKey == null) { + reply = doSendAndReceiveAsyncDefaultCorrelation(requestDestination, jmsRequest, session, priority); + } + else { + reply = doSendAndReceiveAsync(requestDestination, jmsRequest, session, priority); + } + /* + * Remove the gateway's internal correlation Id to avoid conflicts with an upstream + * gateway. + */ + if (reply != null) { + reply.setJMSCorrelationID(null); + } + return reply; + } + finally { + JmsUtils.closeSession(session); + ConnectionFactoryUtils.releaseConnection(connection, this.connectionFactory, true); + } + } + + private javax.jms.Message sendAndReceiveWithoutContainer(Message requestMessage) throws JMSException { Connection connection = this.createConnection(); Session session = null; Destination replyTo = null; @@ -431,7 +668,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { headerMapper.fromHeaders(requestMessage.getHeaders(), jmsRequest); // TODO: support a JmsReplyTo header in the SI Message? - replyTo = this.getReplyDestination(session); + replyTo = this.determineReplyDestination(session); jmsRequest.setJMSReplyTo(replyTo); connection.start(); @@ -515,7 +752,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { 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 " + + "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."); @@ -536,6 +773,100 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { } } + private javax.jms.Message doSendAndReceiveAsync(Destination requestDestination, javax.jms.Message jmsRequest, Session session, int priority) throws JMSException { + String correlationId = null; + MessageProducer messageProducer = null; + try { + messageProducer = session.createProducer(requestDestination); + correlationId = this.gatewayCorrelation + "_" + Long.toString(this.correlationId.incrementAndGet()); + if (this.correlationKey.equals("JMSCorrelationID")) { + jmsRequest.setJMSCorrelationID(correlationId); + } + else { + jmsRequest.setStringProperty(this.correlationKey, correlationId); + } + LinkedBlockingQueue replyQueue = new LinkedBlockingQueue(1); + if (logger.isDebugEnabled()) { + logger.debug(this.getComponentName() + " Sending message with correlationId " + correlationId); + } + this.replies.put(correlationId, replyQueue); + + this.sendRequestMessage(jmsRequest, messageProducer, priority); + + return obtainReplyFromContainer(correlationId, replyQueue); + } + finally { + JmsUtils.closeMessageProducer(messageProducer); + this.replies.remove(correlationId); + } + } + + private javax.jms.Message doSendAndReceiveAsyncDefaultCorrelation(Destination requestDestination, + javax.jms.Message jmsRequest, Session session, int priority) throws JMSException { + String correlationId = null; + MessageProducer messageProducer = null; + + try { + messageProducer = session.createProducer(requestDestination); + LinkedBlockingQueue replyQueue = new LinkedBlockingQueue(1); + + this.sendRequestMessage(jmsRequest, messageProducer, priority); + + correlationId = jmsRequest.getJMSMessageID(); + + if (logger.isDebugEnabled()) { + logger.debug(this.getComponentName() + " Sent message with correlationId " + correlationId); + } + this.replies.put(correlationId, replyQueue); + + /* + * Check to see if the reply arrived before we obtained the correlationId + */ + synchronized (this.earlyOrLateReplies) { + TimedReply timedReply = this.earlyOrLateReplies.remove(correlationId); + if (timedReply != null) { + if (logger.isDebugEnabled()) { + logger.debug("Found early reply with correlationId " + correlationId); + } + replyQueue.add(timedReply.getReply()); + } + } + + return obtainReplyFromContainer(correlationId, replyQueue); + } + finally { + JmsUtils.closeMessageProducer(messageProducer); + this.replies.remove(correlationId); + } + } + + private javax.jms.Message obtainReplyFromContainer(String correlationId, + LinkedBlockingQueue replyQueue) { + javax.jms.Message reply = null; + + if (this.receiveTimeout < 0) { + reply = replyQueue.poll(); + } + else { + try { + reply = replyQueue.poll(this.receiveTimeout, TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + logger.error("Interrupted while awaiting reply; treated as a timeout", e); + Thread.currentThread().interrupt(); + } + } + if (logger.isDebugEnabled()) { + if (reply == null) { + logger.debug(this.getComponentName() + " Timed out waiting for reply with CorrelationId " + correlationId); + } + else { + logger.debug(this.getComponentName() + " Obtained reply with CorrelationId " + correlationId); + } + } + return reply; + } + private void sendRequestMessage(javax.jms.Message jmsRequest, MessageProducer messageProducer, int priority) throws JMSException { if (this.explicitQosEnabled) { messageProducer.send(jmsRequest, this.deliveryMode, priority, this.timeToLive); @@ -582,4 +913,267 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { return connection.createSession(false, Session.AUTO_ACKNOWLEDGE); } + public void onMessage(javax.jms.Message message) { + String correlationId = null; + try { + if (logger.isTraceEnabled()) { + logger.trace(this.getComponentName() + " Received " + message); + } + if (this.correlationKey == null || this.correlationKey.equals("JMSCorrelationID")) { + correlationId = message.getJMSCorrelationID(); + } + else { + correlationId = message.getStringProperty(this.correlationKey); + } + Assert.state(correlationId != null, "Message with no correlationId received"); + LinkedBlockingQueue queue = this.replies.get(correlationId); + if (queue == null) { + if (this.correlationKey != null) { + throw new RuntimeException("No sender waiting for reply"); + } + synchronized (this.earlyOrLateReplies) { + queue = this.replies.get(correlationId); + if (queue == null) { + if (logger.isDebugEnabled()) { + logger.debug("Reply for correlationId " + correlationId + " received early or late"); + } + this.earlyOrLateReplies.put(correlationId, new TimedReply(message)); + } + } + } + if (queue != null) { + if (logger.isDebugEnabled()) { + logger.debug("Received reply with correlationId " + correlationId); + } + queue.add(message); + } + } + catch (Exception e) { + if (logger.isWarnEnabled()) { + logger.warn("Failed to consume reply with correlationId " + correlationId, e); + } + } + } + + private class GatewayReplyListenerContainer extends DefaultMessageListenerContainer { + + private Destination replyDestination; + + @Override + protected Destination resolveDestinationName(Session session, String destinationName) throws JMSException { + if (!StringUtils.hasText(destinationName)) { + this.replyDestination = session.createTemporaryQueue(); + } + else { + this.replyDestination = super.resolveDestinationName(session, destinationName); + } + return this.replyDestination; + } + + + @Override + protected void validateConfiguration() { + if (isSubscriptionDurable() && !isPubSubDomain()) { + throw new IllegalArgumentException("A durable subscription requires a topic (pub-sub domain)"); + } + synchronized (this.lifecycleMonitor) { + if (isSubscriptionDurable() && this.getConcurrentConsumers() != 1) { + throw new IllegalArgumentException("Only 1 concurrent consumer supported for durable subscription"); + } + } + } + + public Destination getReplyDestination() { + Destination replyDest = this.getDestination(); + if (replyDest == null) { + replyDest = this.replyDestination; + } + if (replyDest != null) { + return replyDest; + } + else { + int n = 0; + while (this.replyDestination == null && n++ < 10) { + logger.debug("Waiting for container to create destination"); + try { + Thread.sleep(1000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Container did not establish a destination"); + } + } + if (this.replyDestination == null) { + throw new IllegalStateException("Container did not establish a destination"); + } + else { + return this.replyDestination; + } + } + } + + @Override + protected String getDestinationDescription() { + if (this.replyDestination instanceof TemporaryQueue) { + return "Temporary queue:" + this.replyDestination.toString(); + } + else { + return super.getDestinationDescription(); + } + } + + @Override + protected void recoverAfterListenerSetupFailure() { + this.replyDestination = null; + super.recoverAfterListenerSetupFailure(); + } + } + + private class TimedReply { + + private long timeStamp = System.currentTimeMillis(); + + private javax.jms.Message reply; + + public TimedReply(javax.jms.Message reply) { + this.reply = reply; + } + + public long getTimeStamp() { + return timeStamp; + } + + public javax.jms.Message getReply() { + return reply; + } + } + + private class LateReplyReaper implements Runnable { + + public void run() { + if (logger.isTraceEnabled()) { + logger.trace("Running late reply reaper"); + } + Iterator> lateReplyIterator = earlyOrLateReplies.entrySet().iterator(); + long now = System.currentTimeMillis(); + long expired = now - (JmsOutboundGateway.this.receiveTimeout * 2); + while (lateReplyIterator.hasNext()) { + Entry entry = lateReplyIterator.next(); + if (entry.getValue().getTimeStamp() < expired) { + if (logger.isDebugEnabled()) { + logger.debug("Removing late reply for correlationId " + entry.getKey()); + } + lateReplyIterator.remove(); + } + } + // reschedule myself + if (JmsOutboundGateway.this.receiveTimeout >= 0) { + JmsOutboundGateway.this.reaper = getTaskScheduler().schedule(this, + new Date(now + JmsOutboundGateway.this.receiveTimeout)); + } + } + } + + public static class ReplyContainerProperties { + + private volatile Boolean sessionTransacted; + + private volatile Integer sessionAcknowledgeMode; + + private volatile Long receiveTimeout; + + private volatile Long recoveryInterval; + + private volatile Integer cacheLevel; + + private volatile Integer concurrentConsumers; + + private volatile Integer maxConcurrentConsumers; + + private volatile Integer maxMessagesPerTask; + + private volatile Integer idleConsumerLimit; + + private volatile Integer idleTaskExecutionLimit; + + public Boolean isSessionTransacted() { + return sessionTransacted; + } + + public void setSessionTransacted(Boolean sessionTransacted) { + this.sessionTransacted = sessionTransacted; + } + + public Integer getSessionAcknowledgeMode() { + return sessionAcknowledgeMode; + } + + public void setSessionAcknowledgeMode(Integer sessionAcknowledgeMode) { + this.sessionAcknowledgeMode = sessionAcknowledgeMode; + } + + public Long getReceiveTimeout() { + return receiveTimeout; + } + + public void setReceiveTimeout(Long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + public Long getRecoveryInterval() { + return recoveryInterval; + } + + public void setRecoveryInterval(Long recoveryInterval) { + this.recoveryInterval = recoveryInterval; + } + + public Integer getCacheLevel() { + return cacheLevel; + } + + public void setCacheLevel(Integer cacheLevel) { + this.cacheLevel = cacheLevel; + } + + public Integer getConcurrentConsumers() { + return concurrentConsumers; + } + + public void setConcurrentConsumers(Integer concurrentConsumers) { + this.concurrentConsumers = concurrentConsumers; + } + + public Integer getMaxConcurrentConsumers() { + return maxConcurrentConsumers; + } + + public void setMaxConcurrentConsumers(Integer maxConcurrentConsumers) { + this.maxConcurrentConsumers = maxConcurrentConsumers; + } + + public Integer getMaxMessagesPerTask() { + return maxMessagesPerTask; + } + + public void setMaxMessagesPerTask(Integer maxMessagesPerTask) { + this.maxMessagesPerTask = maxMessagesPerTask; + } + + public Integer getIdleConsumerLimit() { + return idleConsumerLimit; + } + + public void setIdleConsumerLimit(Integer idleConsumerLimit) { + this.idleConsumerLimit = idleConsumerLimit; + } + + public Integer getIdleTaskExecutionLimit() { + return idleTaskExecutionLimit; + } + + public void setIdleTaskExecutionLimit(Integer idleTaskExecutionLimit) { + this.idleTaskExecutionLimit = idleTaskExecutionLimit; + } + } } 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 77ffbb1b60..d8cbfadf54 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-2011 the original author or authors. + * Copyright 2002-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,20 +16,22 @@ package org.springframework.integration.jms.config; -import org.w3c.dom.Element; - import org.springframework.beans.factory.support.BeanDefinitionBuilder; 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; +import org.springframework.util.xml.DomUtils; +import org.w3c.dom.Element; /** * Parser for the <outbound-gateway> element of the integration 'jms' namespace. * * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser { @@ -40,8 +42,7 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser { @Override protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( - "org.springframework.integration.jms.JmsOutboundGateway"); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsOutboundGateway.class); builder.addPropertyReference("connectionFactory", element.getAttribute("connection-factory")); String requestDestination = element.getAttribute("request-destination"); String requestDestinationName = element.getAttribute("request-destination-name"); @@ -96,7 +97,34 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser { else if (StringUtils.hasText(deliveryPersistent)) { builder.addPropertyValue("deliveryPersistent", deliveryPersistent); } + Element container = DomUtils.getChildElementByTagName(element, "reply-listener"); + if (container != null) { + this.parseReplyContainer(builder, parserContext, container); + } return builder; } + private void parseReplyContainer(BeanDefinitionBuilder gatewayBuilder, ParserContext parserContext, Element element) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsOutboundGateway.ReplyContainerProperties.class); + Integer acknowledgeMode = JmsAdapterParserUtils.parseAcknowledgeMode(element, parserContext); + if (acknowledgeMode != null) { + if (acknowledgeMode.intValue() == JmsAdapterParserUtils.SESSION_TRANSACTED) { + builder.addPropertyValue("sessionTransacted", Boolean.TRUE); + } + else { + builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode); + } + } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "concurrent-consumers"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-concurrent-consumers"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-messages-per-task"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "recovery-interval"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "idle-consumer-limit"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "idle-task-execution-limit"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "cache-level"); + + gatewayBuilder.addPropertyValue("replyContainerProperties", builder.getBeanDefinition()); + } + } 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 5c53c83dfe..f6b12aa627 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 @@ -731,9 +731,37 @@ - - + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java new file mode 100644 index 0000000000..57432c7887 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayConnectionTests.java @@ -0,0 +1,149 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.jms; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.Session; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.command.ActiveMQQueue; +import org.junit.Ignore; +import org.junit.Test; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.message.GenericMessage; +import org.springframework.jms.connection.CachingConnectionFactory; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.core.MessageCreator; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + +/** + * @author Gary Russell + * @since 2.2 + * + */ +public class OutboundGatewayConnectionTests { + + private Destination requestQueue1 = new ActiveMQQueue("request1"); + + private Destination replyQueue1 = new ActiveMQQueue("reply1"); + + @Test @Ignore // need a more reliable stop/start for AMQ + public void testContainerWithDestBrokenConnection() throws Exception { + BeanFactory beanFactory = mock(BeanFactory.class); + when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) + .thenReturn(scheduler); + final JmsOutboundGateway gateway = new JmsOutboundGateway(); + gateway.setBeanFactory(beanFactory); + BrokerService broker = new BrokerService(); + broker.addConnector("tcp://localhost:61616?broker.persistent=false"); + broker.start(); + ActiveMQConnectionFactory amqConnectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616"); + ConnectionFactory connectionFactory = new CachingConnectionFactory(amqConnectionFactory); + gateway.setConnectionFactory(connectionFactory); + gateway.setRequestDestination(requestQueue1); + gateway.setReplyDestination(replyQueue1); + gateway.setCorrelationKey("JMSCorrelationID"); + gateway.setUseReplyContainer(true); + gateway.afterPropertiesSet(); + gateway.start(); + final AtomicReference reply = new AtomicReference(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch2.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + JmsTemplate template = new JmsTemplate(); + template.setConnectionFactory(amqConnectionFactory); + template.setReceiveTimeout(5000); + javax.jms.Message request = template.receive(requestQueue1); + assertNotNull(request); + final javax.jms.Message jmsReply = request; + template.send(request.getJMSReplyTo(), new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + return jmsReply; + } + }); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + assertNotNull(reply.get()); + + broker.stop(); + broker = new BrokerService(); + broker.addConnector("tcp://localhost:61616?broker.persistent=false"); + broker.start(); + + final CountDownLatch latch3 = new CountDownLatch(1); + final CountDownLatch latch4 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + latch3.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch4.countDown(); + } + } + }); + assertTrue(latch3.await(10, TimeUnit.SECONDS)); + template = new JmsTemplate(); + template.setConnectionFactory(amqConnectionFactory); + template.setReceiveTimeout(5000); + request = template.receive(requestQueue1); + assertNotNull(request); + final javax.jms.Message jmsReply2 = request; + template.send(request.getJMSReplyTo(), new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + return jmsReply2; + } + }); + assertTrue(latch4.await(10, TimeUnit.SECONDS)); + assertNotNull(reply.get()); + + broker.stop(); + gateway.stop(); + } + +} diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java new file mode 100644 index 0000000000..15149443ad --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/OutboundGatewayFunctionTests.java @@ -0,0 +1,385 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.jms; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.Session; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.command.ActiveMQQueue; +import org.junit.Test; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.message.GenericMessage; +import org.springframework.jms.connection.CachingConnectionFactory; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.core.MessageCreator; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + +/** + * @author Gary Russell + * @since 2.2 + * + */ +public class OutboundGatewayFunctionTests { + + private static Destination requestQueue1 = new ActiveMQQueue("request1"); + + private static Destination replyQueue1 = new ActiveMQQueue("reply1"); + + private static Destination requestQueue2 = new ActiveMQQueue("request2"); + + private static Destination replyQueue2 = new ActiveMQQueue("reply2"); + + private static Destination requestQueue3 = new ActiveMQQueue("request3"); + + private static Destination replyQueue3 = new ActiveMQQueue("reply3"); + + private static Destination requestQueue4 = new ActiveMQQueue("request4"); + + private static Destination replyQueue4 = new ActiveMQQueue("reply4"); + + private static Destination requestQueue5 = new ActiveMQQueue("request5"); + + private static Destination replyQueue5 = new ActiveMQQueue("reply5"); + + private static Destination requestQueue6 = new ActiveMQQueue("request6"); + + private static Destination replyQueue6 = new ActiveMQQueue("reply6"); + + @Test + public void testContainerWithDest() throws Exception { + BeanFactory beanFactory = mock(BeanFactory.class); + when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) + .thenReturn(scheduler); + final JmsOutboundGateway gateway = new JmsOutboundGateway(); + gateway.setBeanFactory(beanFactory); + gateway.setConnectionFactory(getGatewayConnectionFactory()); + gateway.setRequestDestination(requestQueue1); + gateway.setReplyDestination(replyQueue1); + gateway.setCorrelationKey("JMSCorrelationID"); + gateway.setUseReplyContainer(true); + gateway.afterPropertiesSet(); + gateway.start(); + final AtomicReference reply = new AtomicReference(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch2.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + JmsTemplate template = new JmsTemplate(); + template.setConnectionFactory(getTemplateConnectionFactory()); + template.setReceiveTimeout(5000); + javax.jms.Message request = template.receive(requestQueue1); + assertNotNull(request); + final javax.jms.Message jmsReply = request; + template.send(request.getJMSReplyTo(), new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + return jmsReply; + } + }); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + assertNotNull(reply.get()); + + gateway.stop(); + } + + @Test + public void testContainerWithDestNoCorrelation() throws Exception { + BeanFactory beanFactory = mock(BeanFactory.class); + when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) + .thenReturn(scheduler); + final JmsOutboundGateway gateway = new JmsOutboundGateway(); + gateway.setBeanFactory(beanFactory); + gateway.setConnectionFactory(getGatewayConnectionFactory()); + gateway.setRequestDestination(requestQueue2); + gateway.setReplyDestination(replyQueue2); + gateway.setUseReplyContainer(true); + gateway.afterPropertiesSet(); + gateway.start(); + final AtomicReference reply = new AtomicReference(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch2.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + JmsTemplate template = new JmsTemplate(); + template.setConnectionFactory(getTemplateConnectionFactory()); + template.setReceiveTimeout(5000); + javax.jms.Message request = template.receive(requestQueue2); + assertNotNull(request); + final javax.jms.Message jmsReply = request; + template.send(request.getJMSReplyTo(), new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID()); + return jmsReply; + } + }); + assertTrue(latch2.await(20, TimeUnit.SECONDS)); + assertNotNull(reply.get()); + + gateway.stop(); + } + + @Test + public void testContainerWithDestName() throws Exception { + BeanFactory beanFactory = mock(BeanFactory.class); + when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) + .thenReturn(scheduler); + final JmsOutboundGateway gateway = new JmsOutboundGateway(); + gateway.setBeanFactory(beanFactory); + gateway.setConnectionFactory(getGatewayConnectionFactory()); + gateway.setRequestDestination(requestQueue3); + gateway.setReplyDestinationName("reply3"); + gateway.setCorrelationKey("JMSCorrelationID"); + gateway.setUseReplyContainer(true); + gateway.afterPropertiesSet(); + gateway.start(); + final AtomicReference reply = new AtomicReference(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch2.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + JmsTemplate template = new JmsTemplate(); + template.setConnectionFactory(getTemplateConnectionFactory()); + template.setReceiveTimeout(5000); + javax.jms.Message request = template.receive(requestQueue3); + assertNotNull(request); + final javax.jms.Message jmsReply = request; + template.send(request.getJMSReplyTo(), new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + return jmsReply; + } + }); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + assertNotNull(reply.get()); + + gateway.stop(); + } + + @Test + public void testContainerWithDestNameNoCorrelation() throws Exception { + BeanFactory beanFactory = mock(BeanFactory.class); + when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) + .thenReturn(scheduler); + final JmsOutboundGateway gateway = new JmsOutboundGateway(); + gateway.setBeanFactory(beanFactory); + gateway.setConnectionFactory(getGatewayConnectionFactory()); + gateway.setRequestDestination(requestQueue4); + gateway.setReplyDestinationName("reply4"); + gateway.setUseReplyContainer(true); + gateway.afterPropertiesSet(); + gateway.start(); + final AtomicReference reply = new AtomicReference(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch2.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + JmsTemplate template = new JmsTemplate(); + template.setConnectionFactory(getTemplateConnectionFactory()); + template.setReceiveTimeout(5000); + javax.jms.Message request = template.receive(requestQueue4); + assertNotNull(request); + final javax.jms.Message jmsReply = request; + template.send(request.getJMSReplyTo(), new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID()); + return jmsReply; + } + }); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + assertNotNull(reply.get()); + + gateway.stop(); + } + + @Test + public void testContainerWithTemporary() throws Exception { + BeanFactory beanFactory = mock(BeanFactory.class); + when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) + .thenReturn(scheduler); + final JmsOutboundGateway gateway = new JmsOutboundGateway(); + gateway.setBeanFactory(beanFactory); + gateway.setConnectionFactory(getGatewayConnectionFactory()); + gateway.setRequestDestination(requestQueue5); + gateway.setCorrelationKey("JMSCorrelationID"); + gateway.setUseReplyContainer(true); + gateway.afterPropertiesSet(); + gateway.start(); + final AtomicReference reply = new AtomicReference(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch2.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + JmsTemplate template = new JmsTemplate(); + template.setConnectionFactory(getTemplateConnectionFactory()); + template.setReceiveTimeout(5000); + javax.jms.Message request = template.receive(requestQueue5); + assertNotNull(request); + final javax.jms.Message jmsReply = request; + template.send(request.getJMSReplyTo(), new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + return jmsReply; + } + }); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + assertNotNull(reply.get()); + + gateway.stop(); + } + + @Test + public void testContainerWithTemporaryNoCorrelation() throws Exception { + BeanFactory beanFactory = mock(BeanFactory.class); + when(beanFactory.containsBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)).thenReturn(true); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.initialize(); + when(beanFactory.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)) + .thenReturn(scheduler); + final JmsOutboundGateway gateway = new JmsOutboundGateway(); + gateway.setBeanFactory(beanFactory); + gateway.setConnectionFactory(getGatewayConnectionFactory()); + gateway.setRequestDestination(requestQueue6); + gateway.setUseReplyContainer(true); + gateway.afterPropertiesSet(); + gateway.start(); + final AtomicReference reply = new AtomicReference(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + latch1.countDown(); + try { + reply.set(gateway.handleRequestMessage(new GenericMessage("foo"))); + } + finally { + latch2.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + JmsTemplate template = new JmsTemplate(); + template.setConnectionFactory(getTemplateConnectionFactory()); + template.setReceiveTimeout(5000); + javax.jms.Message request = template.receive(requestQueue6); + assertNotNull(request); + final javax.jms.Message jmsReply = request; + template.send(request.getJMSReplyTo(), new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + jmsReply.setJMSCorrelationID(jmsReply.getJMSMessageID()); + return jmsReply; + } + }); + assertTrue(latch2.await(10, TimeUnit.SECONDS)); + assertNotNull(reply.get()); + + gateway.stop(); + } + + private ConnectionFactory getTemplateConnectionFactory() { + ConnectionFactory amqConnectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"); + return amqConnectionFactory; + } + + private ConnectionFactory getGatewayConnectionFactory() { + ConnectionFactory amqConnectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false"); + return new CachingConnectionFactory(amqConnectionFactory); + } + +} diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java index 4d4ca31b1c..3a387bcc51 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java @@ -46,6 +46,7 @@ import org.springframework.integration.jms.JmsOutboundGateway; import org.springframework.integration.jms.StubMessageConverter; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.test.util.TestUtils; +import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.jms.support.converter.MessageConverter; /** @@ -68,6 +69,17 @@ public class JmsOutboundGatewayParserTests { accessor = new DirectFieldAccessor(gateway); int deliveryMode = (Integer)accessor.getPropertyValue("deliveryMode"); assertEquals(DeliveryMode.PERSISTENT, deliveryMode); + DefaultMessageListenerContainer container = TestUtils.getPropertyValue(gateway, "replyContainer", + DefaultMessageListenerContainer.class); + assertEquals(4, TestUtils.getPropertyValue(container, "concurrentConsumers")); + assertEquals(5, TestUtils.getPropertyValue(container, "maxConcurrentConsumers")); + assertEquals(10, TestUtils.getPropertyValue(container, "maxMessagesPerTask")); + assertEquals(2000L, TestUtils.getPropertyValue(container, "receiveTimeout")); + assertEquals(10000L, TestUtils.getPropertyValue(container, "recoveryInterval")); + assertEquals(7, TestUtils.getPropertyValue(container, "idleConsumerLimit")); + assertEquals(2, TestUtils.getPropertyValue(container, "idleTaskExecutionLimit")); + assertEquals(3, TestUtils.getPropertyValue(container, "cacheLevel")); + assertTrue(container.isSessionTransacted()); } @Test @@ -78,6 +90,7 @@ public class JmsOutboundGatewayParserTests { JmsOutboundGateway gateway = TestUtils.getPropertyValue(endpoint, "handler", JmsOutboundGateway.class); gateway.handleMessage(new GenericMessage("foo")); assertEquals(1, adviceCalled); + assertEquals(3, TestUtils.getPropertyValue(gateway, "replyContainer.sessionAcknowledgeMode")); } @Test diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsPriorityTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsPriorityTests.java index ab453efe94..43063c67a4 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsPriorityTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsPriorityTests.java @@ -28,7 +28,6 @@ import javax.jms.TextMessage; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; - import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; @@ -111,6 +110,7 @@ public class JmsPriorityTests { String text = "priority=" + request.getJMSPriority(); TextMessage reply = session.createTextMessage(text); MessageProducer producer = session.createProducer(request.getJMSReplyTo()); + reply.setJMSCorrelationID(request.getJMSMessageID()); producer.send(reply); } } 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 7682e03a10..c2d3bd411d 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,7 +58,6 @@ public class JmsWithMarshallingMessageConverterTests { MessageHeaders headers = replyMessage.getHeaders(); // check for couple of JMS headers, make sure they are present assertNotNull(headers.get("jms_redelivered")); - assertNotNull(headers.get("jms_correlationId")); assertEquals("HELLO", replyMessage.getPayload()); } @@ -86,7 +85,7 @@ public class JmsWithMarshallingMessageConverterTests { public boolean supports(Class clazz) { return true; - } + } } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithDeliveryPersistent.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithDeliveryPersistent.xml index b6f8e6074e..7ad15d5012 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithDeliveryPersistent.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithDeliveryPersistent.xml @@ -15,8 +15,19 @@ - + delivery-persistent="true"> + + + @@ -32,6 +43,16 @@ + 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 new file mode 100644 index 0000000000..2d356b7a66 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/MiscellaneousTests.java @@ -0,0 +1,72 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.jms.request_reply; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.gateway.RequestReplyExchanger; +import org.springframework.integration.jms.config.ActiveMqTestUtils; +import org.springframework.integration.message.GenericMessage; +import org.springframework.util.StopWatch; +/** + * @author Oleg Zhurakousky + */ +public class MiscellaneousTests { + + /** + * Asserts that receive-timeout is honored even if + * requests (once in process), takes less then receive-timeout value + * when requests are queued up (e.g., single consumer receiver) + */ + @Test + public void testTimeoutHonoringWhenRequestsQueuedUp() throws Exception{ + ActiveMqTestUtils.prepare(); + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("honor-timeout.xml", this.getClass()); + final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class); + final CountDownLatch latch = new CountDownLatch(3); + final AtomicInteger replies = new AtomicInteger(); + StopWatch stopWatch = new StopWatch(); + stopWatch.start(); + for (int i = 0; i < 3; i++) { + this.exchange(latch, gateway, replies); + } + latch.await(); + stopWatch.stop(); + assertTrue(stopWatch.getTotalTimeMillis() <= 12000); + assertEquals(1, replies.get()); + } + + + private void exchange(final CountDownLatch latch, final RequestReplyExchanger gateway, final AtomicInteger replies) { + new Thread(new Runnable() { + public void run() { + try { + gateway.exchange(new GenericMessage("")); + 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 new file mode 100644 index 0000000000..922b0375f5 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineJmsTests.java @@ -0,0 +1,168 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.jms.request_reply; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.MessageTimeoutException; +import org.springframework.integration.gateway.RequestReplyExchanger; +import org.springframework.integration.jms.config.ActiveMqTestUtils; +import org.springframework.integration.message.GenericMessage; +/** + * @author Oleg Zhurakousky + */ +public class PipelineJmsTests { + + private final Executor executor = Executors.newFixedThreadPool(30); + + int requests = 50; + + /** + * jms:out -> jms:in -> randomTimeoutProcess -> + * jms:out -> jms:in + * All reply queues are TEMPORARY + */ + @Test + public void testPipeline1() throws Exception{ + this.test("pipeline-01.xml"); + } + + /** + * jms:out(correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess -> + * jms:out -> jms:in + * All reply queues are TEMPORARY + */ + @Test + public void testPipeline2() throws Exception{ + this.test("pipeline-02.xml"); + } + + /** + * jms:out(correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess -> + * jms:out(correlation-key="JMSCorrelationID") -> jms:in + * All reply queues are TEMPORARY + */ + @Test + public void testPipeline3() throws Exception{ + this.test("pipeline-03.xml"); + } + + /** + * jms:out -> jms:in -> randomTimeoutProcess -> + * jms:out(correlation-key="JMSCorrelationID") -> jms:in + * All reply queues are TEMPORARY + */ + @Test + public void testPipeline4() throws Exception{ + this.test("pipeline-04.xml"); + } + + /** + * jms:out(correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess -> + * jms:out -> jms:in + * All reply queues are TEMPORARY + */ + @Test + public void testPipeline5() throws Exception{ + this.test("pipeline-05.xml"); + } + + /** + * jms:out -> jms:in -> randomTimeoutProcess -> + * jms:out(correlation-key="foo") -> jms:in(correlation-key="foo") + * All reply queues are TEMPORARY + */ + @Test + public void testPipeline6() throws Exception{ + this.test("pipeline-06.xml"); + } + + /** + * jms:out(correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess -> + * jms:out(correlation-key="foo") -> jms:in(correlation-key="foo") + * All reply queues are TEMPORARY + */ + @Test + public void testPipeline7() throws Exception{ + this.test("pipeline-07.xml"); + } + + /** + * jms:out(correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess -> + * jms:out(correlation-key="JMSCorrelationID") -> jms:in + * All reply queues are TEMPORARY + */ + @Test + public void testPipeline8() throws Exception{ + this.test("pipeline-08.xml"); + } + + /** + * jms:out(correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess -> + * jms:out(correlation-key="bar") -> jms:in(correlation-key="bar") + * All reply queues are TEMPORARY + */ + @Test + public void testPipeline9() throws Exception{ + this.test("pipeline-09.xml"); + } + + public void test(String contextConfig) throws Exception{ + ActiveMqTestUtils.prepare(); + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(contextConfig, this.getClass()); + final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class); + final CountDownLatch latch = new CountDownLatch(requests); + final AtomicInteger successCounter = new AtomicInteger(); + final AtomicInteger timeoutCounter = new AtomicInteger(); + final AtomicInteger failureCounter = new AtomicInteger(); + + for (int i = 0; i < requests; i++) { + final int y = i; + executor.execute(new Runnable() { + public void run() { + try { + assertEquals(y, gateway.exchange(new GenericMessage(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 new file mode 100644 index 0000000000..296331115a --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/PipelineNamedReplyQueuesJmsTests.java @@ -0,0 +1,186 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.jms.request_reply; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.MessageTimeoutException; +import org.springframework.integration.gateway.RequestReplyExchanger; +import org.springframework.integration.jms.config.ActiveMqTestUtils; +import org.springframework.integration.message.GenericMessage; +/** + * @author Oleg Zhurakousky + * @author Gary Russell + */ +public class PipelineNamedReplyQueuesJmsTests { + + private final Executor executor = Executors.newFixedThreadPool(30); + + int requests = 50; + + int timeouts; + + /** + * jms:out(reply-destination-name="pipeline01-01") -> jms:in -> randomTimeoutProcess -> + * jms:out -> jms:in + */ + @Test + public void testPipeline1() throws Exception{ + this.test("pipeline-named-queue-01.xml"); + } + + /** + * jms:out(reply-destination-name="pipeline02-01") -> jms:in -> randomTimeoutProcess -> + * jms:out(reply-destination-name="pipeline02-02") -> jms:in -> + * jms:out(reply-destination-name="pipeline02-03") -> jms:in + */ + @Test + public void testPipeline2() throws Exception{ + this.test("pipeline-named-queue-02.xml"); + } + + /** + * Same as {@link #testPipeline2()} except all gateways use the same reply queue. + * and zero failures expected (no timeouts on server). + * jms:out(reply-destination-name="pipeline02a-01") -> jms:in -> zeroTimeoutProcess -> + * jms:out(reply-destination-name="pipeline02a-01") -> jms:in -> + * jms:out(reply-destination-name="pipeline02a-01") -> jms:in + */ + @Test + public void testPipeline2a() throws Exception{ + this.test("pipeline-named-queue-02a.xml"); + assertEquals(0, this.timeouts); + } + + /** + * jms:out(reply-destination-name="pipeline03-01", correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess -> + * jms:out(reply-destination-name="pipeline03-02") -> jms:in -> + * jms:out(reply-destination-name="pipeline03-03") -> jms:in + */ + @Test + public void testPipeline3() throws Exception{ + this.test("pipeline-named-queue-03.xml"); + } + + /** + * Same as {@link #testPipeline3()} except all gateways use the same reply queue. + * Ensures the correlation id is not propagated. No timeouts expected. + * jms:out(reply-destination-name="pipeline03a-01", correlation-key="JMSCorrelationID") -> jms:in -> zeroTimeoutProcess -> + * jms:out(reply-destination-name="pipeline03a-01") -> jms:in -> + * jms:out(reply-destination-name="pipeline03a-01") -> jms:in + * Ensures reply came from service after third gateway + */ + @Test + public void testPipeline3a() throws Exception{ + this.test("pipeline-named-queue-03a.xml", 20000); + assertEquals(0, this.timeouts); + } + + /** + * jms:out(reply-destination-name="pipeline04-01", correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess -> + * jms:out(reply-destination-name="pipeline04-02") -> jms:in + */ + @Test + public void testPipeline4() throws Exception{ + this.test("pipeline-named-queue-04.xml"); + } + + /** + * jms:out(reply-destination-name="pipeline05-01", correlation-key="foo") -> jms:in(correlation-key="foo") -> randomTimeoutProcess -> + * jms:out(reply-destination-name="pipeline05-02", correlation-key="JMSCorrelationID") -> jms:in + */ + @Test + public void testPipeline5() throws Exception{ + this.test("pipeline-named-queue-05.xml"); + } + + /** + * jms:out(reply-destination-name="pipeline06-01", correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess -> + * jms:out(reply-destination-name="pipeline06-02", correlation-key="foo") -> jms:in(correlation-key="foo") + */ + @Test + public void testPipeline6() throws Exception{ + this.test("pipeline-named-queue-06.xml"); + } + + /** + * jms:out(reply-destination-name="pipeline07-01", correlation-key="JMSCorrelationID") -> jms:in -> randomTimeoutProcess -> + * jms:out(reply-destination-name="pipeline07-02", correlation-key="foo") -> jms:in(correlation-key="foo") + */ + @Test + public void testPipeline7() throws Exception{ + this.test("pipeline-named-queue-07.xml"); + } + + public void test(String contextConfig) throws Exception { + test(contextConfig, 0); + } + + public void test(String contextConfig, final int offset) throws Exception { + this.timeouts = 0; + ActiveMqTestUtils.prepare(); + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(contextConfig, this.getClass()); + final AtomicInteger successCounter = new AtomicInteger(); + final AtomicInteger timeoutCounter = new AtomicInteger(); + final AtomicInteger failureCounter = new AtomicInteger(); + try { + final RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class); + final CountDownLatch latch = new CountDownLatch(requests); + + for (int i = 0; i < requests; i++) { + final int y = i; + executor.execute(new Runnable() { + public void run() { + try { + assertEquals(y + offset, gateway.exchange(new GenericMessage(y)).getPayload()); + successCounter.incrementAndGet(); + } catch (MessageTimeoutException e) { + timeoutCounter.incrementAndGet(); + } catch (Throwable t) { + t.printStackTrace(); + failureCounter.incrementAndGet(); + } finally { + latch.countDown(); + } + } + }); + } + assertTrue(latch.await(60, TimeUnit.SECONDS)); + // technically all we care that its > 0, + // but reality of this test it has to be something more then 0 + assertTrue(successCounter.get() > 10); + assertEquals(0, failureCounter.get()); + assertEquals(requests, successCounter.get() + timeoutCounter.get()); + this.timeouts = timeoutCounter.get(); + } + finally { + System.out.println(contextConfig); + System.out.println("Success: " + successCounter.get()); + System.out.println("Timeout: " + timeoutCounter.get()); + System.out.println("Failure: " + failureCounter.get()); + context.destroy(); + } + } +} 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 new file mode 100644 index 0000000000..d8ba775b31 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCachedConsumersTests.java @@ -0,0 +1,265 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.jms.request_reply; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import java.util.concurrent.CountDownLatch; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageProducer; +import javax.jms.Session; +import javax.jms.TextMessage; + +import org.junit.Test; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.MessageTimeoutException; +import org.springframework.integration.gateway.RequestReplyExchanger; +import org.springframework.integration.jms.JmsOutboundGateway; +import org.springframework.integration.jms.config.ActiveMqTestUtils; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.jms.connection.CachingConnectionFactory; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.core.MessageCreator; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.jms.listener.SessionAwareMessageListener; +import org.springframework.jms.support.converter.SimpleMessageConverter; +/** + * @author Oleg Zhurakousky + */ +public class RequestReplyScenariosWithCachedConsumersTests { + + private final SimpleMessageConverter converter = new SimpleMessageConverter(); + + @Test(expected=MessageTimeoutException.class) + public void messageCorrelationBasedOnRequestMessageIdOptimized() throws Exception{ + ActiveMqTestUtils.prepare(); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass()); + try { + RequestReplyExchanger gateway = context.getBean("standardMessageIdCopyingConsumerWithOptimization", RequestReplyExchanger.class); + CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class); + final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + + + final Destination requestDestination = context.getBean("siOutQueueOptimizedA", Destination.class); + final Destination replyDestination = context.getBean("siInQueueOptimizedA", Destination.class); + new Thread(new Runnable() { + + public void run() { + final Message requestMessage = jmsTemplate.receive(requestDestination); + jmsTemplate.send(replyDestination, new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + TextMessage message = session.createTextMessage(); + message.setText("bar"); + message.setJMSCorrelationID(requestMessage.getJMSMessageID()); + return message; + } + }); + } + }).start(); + gateway.exchange(new GenericMessage("foo")); + } + finally { + context.destroy(); + } + + } + + @Test + public void messageCorrelationBasedOnRequestMessageIdNonOptimized() throws Exception{ + ActiveMqTestUtils.prepare(); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass()); + try { + RequestReplyExchanger gateway = context.getBean("standardMessageIdCopyingConsumerWithoutOptimization", RequestReplyExchanger.class); + CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class); + final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + + + final Destination requestDestination = context.getBean("siOutQueueNonOptimizedB", Destination.class); + final Destination replyDestination = context.getBean("siInQueueNonOptimizedB", Destination.class); + new Thread(new Runnable() { + + public void run() { + final Message requestMessage = jmsTemplate.receive(requestDestination); + jmsTemplate.send(replyDestination, new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + TextMessage message = session.createTextMessage(); + message.setText("bar"); + message.setJMSCorrelationID(requestMessage.getJMSMessageID()); + return message; + } + }); + } + }).start(); + org.springframework.integration.Message siReplyMessage = gateway.exchange(new GenericMessage("foo")); + assertEquals("bar", siReplyMessage.getPayload()); + } + finally { + context.destroy(); + } + } + + @Test + public void messageCorrelationBasedOnRequestCorrelationIdOptimized() throws Exception{ + ActiveMqTestUtils.prepare(); + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass()); + try { + RequestReplyExchanger gateway = context.getBean("correlationPropagatingConsumerWithOptimization", RequestReplyExchanger.class); + CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class); + + final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + + final Destination requestDestination = context.getBean("siOutQueueOptimizedC", Destination.class); + final Destination replyDestination = context.getBean("siInQueueOptimizedC", Destination.class); + new Thread(new Runnable() { + + public void run() { + final Message requestMessage = jmsTemplate.receive(requestDestination); + jmsTemplate.send(replyDestination, new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + TextMessage message = session.createTextMessage(); + message.setText("bar"); + message.setJMSCorrelationID(requestMessage.getJMSCorrelationID()); + return message; + } + }); + } + }).start(); + org.springframework.integration.Message siReplyMessage = gateway.exchange(new GenericMessage("foo")); + assertEquals("bar", siReplyMessage.getPayload()); + } + finally { + context.destroy(); + } + } + + @Test(expected=MessageTimeoutException.class) + public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() throws Exception{ + ActiveMqTestUtils.prepare(); + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass()); + try { + RequestReplyExchanger gateway = context.getBean("correlationPropagatingConsumerWithoutOptimization", RequestReplyExchanger.class); + CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class); + + final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + + final Destination requestDestination = context.getBean("siOutQueueNonOptimizedD", Destination.class); + final Destination replyDestination = context.getBean("siInQueueNonOptimizedD", Destination.class); + new Thread(new Runnable() { + + public void run() { + final Message requestMessage = jmsTemplate.receive(requestDestination); + jmsTemplate.send(replyDestination, new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + TextMessage message = session.createTextMessage(); + message.setText("bar"); + message.setJMSCorrelationID(requestMessage.getJMSCorrelationID()); + return message; + } + }); + } + }).start(); + org.springframework.integration.Message siReplyMessage = gateway.exchange(new GenericMessage("foo")); + assertEquals("bar", siReplyMessage.getPayload()); + } + finally { + context.destroy(); + } + } + + @Test + public void messageCorrelationBasedOnRequestCorrelationIdTimedOutFirstReplyOptimized() throws Exception{ + ActiveMqTestUtils.prepare(); + ClassPathXmlApplicationContext context = + new ClassPathXmlApplicationContext("producer-cached-consumers.xml", this.getClass()); + try { + RequestReplyExchanger gateway = + context.getBean("correlationPropagatingConsumerWithOptimizationDelayFirstReply", RequestReplyExchanger.class); + final ConnectionFactory connectionFactory = context.getBean("connectionFactory", ConnectionFactory.class); + + final Destination requestDestination = context.getBean("siOutQueueE", Destination.class); + final Destination replyDestination = context.getBean("siInQueueE", Destination.class); + + for (int i = 0; i < 3; i++) { + System.out.println("#### " + i); + try { + gateway.exchange(gateway.exchange(new GenericMessage("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()); + } + finally { + context.destroy(); + } + } + + 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 new file mode 100644 index 0000000000..1a9c34ae06 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithCorrelationKeyProvidedTests.java @@ -0,0 +1,84 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.jms.request_reply; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; + +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.gateway.RequestReplyExchanger; +import org.springframework.integration.jms.JmsOutboundGateway; +import org.springframework.integration.jms.config.ActiveMqTestUtils; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; +/** + * @author Oleg Zhurakousky + */ +public class RequestReplyScenariosWithCorrelationKeyProvidedTests { + + @Test + public void messageCorrelationBasedCustomCorrelationKey() throws Exception{ + ActiveMqTestUtils.prepare(); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass()); + RequestReplyExchanger gateway = context.getBean("explicitCorrelationKeyGateway", RequestReplyExchanger.class); + + gateway.exchange(MessageBuilder.withPayload("foo").build()); + context.close(); + } + + @Test + public void messageCorrelationBasedCustomCorrelationKeyAsJMSCorrelationID() throws Exception{ + ActiveMqTestUtils.prepare(); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass()); + RequestReplyExchanger gateway = context.getBean("explicitCorrelationKeyGatewayB", RequestReplyExchanger.class); + + gateway.exchange(MessageBuilder.withPayload("foo").build()); + context.close(); + } + + @Test + public void messageCorrelationBasedCustomCorrelationKeyDelayedReplies() throws Exception{ + ActiveMqTestUtils.prepare(); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("explicit-correlation-key.xml", this.getClass()); + RequestReplyExchanger gateway = context.getBean("explicitCorrelationKeyGatewayC", RequestReplyExchanger.class); + + + for (int i = 0; i < 3; i++) { + try { + gateway.exchange(MessageBuilder.withPayload("hello").build()); + } catch (Exception e) { + // ignore + } + } + + JmsOutboundGateway outGateway = TestUtils.getPropertyValue(context.getBean("outGateway"), "handler", JmsOutboundGateway.class); + outGateway.setReceiveTimeout(5000); + assertEquals("foo", gateway.exchange(MessageBuilder.withPayload("foo").build()).getPayload()); + context.close(); + } + + + public static class DelayedService { + public String echo(String s) throws Exception{ + Thread.sleep(200); + return s; + } + } +} 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 new file mode 100644 index 0000000000..981880a978 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithNonCachedConsumersTests.java @@ -0,0 +1,175 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.jms.request_reply; + +import static org.junit.Assert.assertEquals; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.Session; +import javax.jms.TextMessage; + +import org.junit.Test; +import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.MessageTimeoutException; +import org.springframework.integration.gateway.RequestReplyExchanger; +import org.springframework.integration.jms.config.ActiveMqTestUtils; +import org.springframework.integration.message.GenericMessage; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.core.MessageCreator; +/** + * @author Oleg Zhurakousky + */ +public class RequestReplyScenariosWithNonCachedConsumersTests { + + @Test(expected=MessageTimeoutException.class) + public void messageCorrelationBasedOnRequestMessageIdOptimized() throws Exception{ + ActiveMqTestUtils.prepare(); + AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass()); + try { + RequestReplyExchanger gateway = context.getBean("optimizedMessageId", RequestReplyExchanger.class); + ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); + final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + + final Destination requestDestination = context.getBean("siOutQueueC", Destination.class); + final Destination replyDestination = context.getBean("siInQueueC", Destination.class); + new Thread(new Runnable() { + + public void run() { + final Message requestMessage = jmsTemplate.receive(requestDestination); + jmsTemplate.send(replyDestination, new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + TextMessage message = session.createTextMessage(); + message.setText("bar"); + message.setJMSCorrelationID(requestMessage.getJMSMessageID()); + return message; + } + }); + } + }).start(); + org.springframework.integration.Message siReplyMessage = gateway.exchange(new GenericMessage("foo")); + assertEquals("bar", siReplyMessage.getPayload()); + } + finally { + context.destroy(); + } + } + + @Test + public void messageCorrelationBasedOnRequestMessageIdNonOptimized() throws Exception{ + ActiveMqTestUtils.prepare(); + AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass()); + try { + RequestReplyExchanger gateway = context.getBean("nonoptimizedMessageId", RequestReplyExchanger.class); + ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); + final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + + final Destination requestDestination = context.getBean("siOutQueueD", Destination.class); + final Destination replyDestination = context.getBean("siInQueueD", Destination.class); + new Thread(new Runnable() { + + public void run() { + final Message requestMessage = jmsTemplate.receive(requestDestination); + jmsTemplate.send(replyDestination, new MessageCreator() { + public Message createMessage(Session session) throws JMSException { + TextMessage message = session.createTextMessage(); + message.setText("bar"); + message.setJMSCorrelationID(requestMessage.getJMSMessageID()); + return message; + } + }); + } + }).start(); + org.springframework.integration.Message siReplyMessage = gateway.exchange(new GenericMessage("foo")); + assertEquals("bar", siReplyMessage.getPayload()); + } + finally { + context.destroy(); + } + } + + @Test + public void messageCorrelationBasedOnRequestCorrelationIdOptimized() throws Exception{ + ActiveMqTestUtils.prepare(); + AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass()); + try { + RequestReplyExchanger gateway = context.getBean("optimized", RequestReplyExchanger.class); + ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); + final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + + final Destination requestDestination = context.getBean("siOutQueueA", Destination.class); + final Destination replyDestination = context.getBean("siInQueueA", Destination.class); + new Thread(new Runnable() { + + public void run() { + final Message requestMessage = jmsTemplate.receive(requestDestination); + jmsTemplate.send(replyDestination, new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + TextMessage message = session.createTextMessage(); + message.setText("bar"); + message.setJMSCorrelationID(requestMessage.getJMSCorrelationID()); + return message; + } + }); + } + }).start(); + org.springframework.integration.Message siReplyMessage = gateway.exchange(new GenericMessage("foo")); + assertEquals("bar", siReplyMessage.getPayload()); + } + finally { + context.destroy(); + } + } + + @Test(expected=MessageTimeoutException.class) + public void messageCorrelationBasedOnRequestCorrelationIdNonOptimized() throws Exception{ + ActiveMqTestUtils.prepare(); + AbstractApplicationContext context = new ClassPathXmlApplicationContext("producer-no-cached-consumers.xml", this.getClass()); + try { + RequestReplyExchanger gateway = context.getBean("nonoptimized", RequestReplyExchanger.class); + ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class); + final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + + final Destination requestDestination = context.getBean("siOutQueueB", Destination.class); + final Destination replyDestination = context.getBean("siInQueueB", Destination.class); + new Thread(new Runnable() { + + public void run() { + final Message requestMessage = jmsTemplate.receive(requestDestination); + jmsTemplate.send(replyDestination, new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + TextMessage message = session.createTextMessage(); + message.setText("bar"); + message.setJMSCorrelationID(requestMessage.getJMSCorrelationID()); + return message; + } + }); + } + }).start(); + org.springframework.integration.Message siReplyMessage = gateway.exchange(new GenericMessage("foo")); + assertEquals("bar", siReplyMessage.getPayload()); + } + finally { + context.destroy(); + } + } +} 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 new file mode 100644 index 0000000000..78909846d0 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/RequestReplyScenariosWithTempReplyQueuesTests.java @@ -0,0 +1,275 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.jms.request_reply; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.List; +import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageProducer; +import javax.jms.Session; +import javax.jms.TextMessage; + +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.command.ActiveMQDestination; +import org.junit.Test; + +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.gateway.RequestReplyExchanger; +import org.springframework.integration.jms.config.ActiveMqTestUtils; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.jms.connection.CachingConnectionFactory; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.core.MessageCreator; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.jms.listener.SessionAwareMessageListener; +import org.springframework.jms.support.converter.SimpleMessageConverter; +/** + * @author Oleg Zhurakousky + */ +public class RequestReplyScenariosWithTempReplyQueuesTests { + + private final SimpleMessageConverter converter = new SimpleMessageConverter(); + + @Test + public void messageCorrelationBasedOnRequestMessageId() throws Exception{ + ActiveMqTestUtils.prepare(); + + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("producer-temp-reply-consumers.xml", this.getClass()); + RequestReplyExchanger gateway = context.getBean(RequestReplyExchanger.class); + CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class); + final JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory); + + final Destination requestDestination = context.getBean("siOutQueue", Destination.class); + + new Thread(new Runnable() { + + public void run() { + final Message requestMessage = jmsTemplate.receive(requestDestination); + Destination replyTo = null; + try { + replyTo = requestMessage.getJMSReplyTo(); + } catch (Exception e) { + fail(); + } + jmsTemplate.send(replyTo, new MessageCreator() { + + public Message createMessage(Session session) throws JMSException { + try { + TextMessage message = session.createTextMessage(); + message.setText("bar"); + message.setJMSCorrelationID(requestMessage.getJMSMessageID()); + return message; + } catch (Exception e) { + // ignore + } + return null; + } + }); + } + }).start(); + gateway.exchange(new GenericMessage("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 new file mode 100644 index 0000000000..219613d57a --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/broken-broker.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..23f68c040e --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/explicit-correlation-key.xml @@ -0,0 +1,82 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..00337c6b17 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/honor-timeout.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..a74d179060 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/mult-producer-and-consumers-temp-reply.xml @@ -0,0 +1,49 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..57894920fb --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-01.xml @@ -0,0 +1,62 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..d9d6d2db7b --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-02.xml @@ -0,0 +1,63 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..7e5cc917e1 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-03.xml @@ -0,0 +1,64 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..9e8a50af7b --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-04.xml @@ -0,0 +1,63 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..23c0a2f55b --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-05.xml @@ -0,0 +1,64 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..b95a0f1e0a --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-06.xml @@ -0,0 +1,64 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..8d81300a32 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-07.xml @@ -0,0 +1,65 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..f978866ea2 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-08.xml @@ -0,0 +1,65 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..5d5cffe455 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-09.xml @@ -0,0 +1,66 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..472f068088 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-01.xml @@ -0,0 +1,63 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..b63ccf0f09 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02.xml @@ -0,0 +1,83 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..e173864b99 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-02a.xml @@ -0,0 +1,85 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..4a6b302020 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03.xml @@ -0,0 +1,84 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..f5fee625de --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-03a.xml @@ -0,0 +1,87 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..2798d15e10 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-04.xml @@ -0,0 +1,66 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..f092a38611 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-05.xml @@ -0,0 +1,67 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..5da8830ece --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-06.xml @@ -0,0 +1,68 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..1be33625fb --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/pipeline-named-queue-07.xml @@ -0,0 +1,68 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..877092dfa1 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-cached-consumers.xml @@ -0,0 +1,118 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..3def60cbbf --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-no-cached-consumers.xml @@ -0,0 +1,95 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..857f6fe238 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/producer-temp-reply-consumers.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jms/src/test/resources/log4j.properties b/spring-integration-jms/src/test/resources/log4j.properties index 0d11d824ba..b8ebc43ebe 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=WARN, stdout +log4j.rootCategory=ERROR, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m%n -log4j.category.org.springframework=WARN -# log4j.category.org.springframework.integration=DEBUG +log4j.category.org.springframework=ERROR +#log4j.category.org.springframework.integration.jms=DEBUG # log4j.category.org.springframework.integration.jdbc=DEBUG -log4j.category.org.springframework.jms=DEBUG +log4j.category.org.springframework.jms=ERROR