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

INT-3944: Async JMS Outbound Gateway

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3944

Polishing (PR comments) and Doc Polish

Polishing - PR Comments

Remove custom `async` boolean in favor of the now public setter.

Add exception to `JmsException` hierarchy.

INT-3944: Polishing

* Rename `asyncReplySupported` just to `async`. As well as its getter and setter. Plus fix docs on the matter
* Polishing for the `JmsOutboundGateway` according PR comments
* Rework `JmsOutboundGateway` to return `AbstractIntegrationMessageBuilder` instead of `Message`
since `AbstractMessageProducingHandler` rebuilds `Message` for a new one to copy headers from request
* Add `getPayload()` and `getHeaders()` to the `AbstractIntegrationMessageBuilder` to make the `Routing Slip`
users happy, when the `AbstractIntegrationMessageBuilder` is pushed to the `Routing Slip path` function
* Fix race condition in the `ChatMessageListeningEndpointTests`: the `stanza` parsing is done in the different Thread.

Doc Polishing (Routing Slip)

Fix timing issue in the `LastModifiedFileListFilterTests`:
`Thread.sleep()` not always reflects the reality.
Switch to the "past simulation" via explicit `File.setLastModified()`
This commit is contained in:
Gary Russell
2016-03-23 10:12:13 -04:00
committed by Artem Bilan
parent 0fd72d2d18
commit 4bfcdb9dfa
24 changed files with 539 additions and 325 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -51,6 +51,7 @@ import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.jms.util.JmsAdapterUtils;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.jms.connection.ConnectionFactoryUtils;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.support.JmsUtils;
@@ -66,6 +67,7 @@ import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.SettableListenableFuture;
/**
* An outbound Messaging Gateway for request/reply JMS.
@@ -79,6 +81,23 @@ import org.springframework.util.StringUtils;
*/
public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler implements Lifecycle, MessageListener {
private final Object initializationMonitor = new Object();
private final AtomicLong correlationId = new AtomicLong();
private final String gatewayCorrelation = UUID.randomUUID().toString();
private final Map<String, LinkedBlockingQueue<javax.jms.Message>> replies =
new ConcurrentHashMap<String, LinkedBlockingQueue<javax.jms.Message>>();
private final ConcurrentHashMap<String, TimedReply> earlyOrLateReplies =
new ConcurrentHashMap<String, JmsOutboundGateway.TimedReply>();
private final Map<String, SettableListenableFuture<AbstractIntegrationMessageBuilder<?>>> futures =
new ConcurrentHashMap<String, SettableListenableFuture<AbstractIntegrationMessageBuilder<?>>>();
private final Object lifeCycleMonitor = new Object();
private volatile Destination requestDestination;
private volatile String requestDestinationName;
@@ -127,24 +146,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
private volatile boolean useReplyContainer;
private final Object initializationMonitor = new Object();
private volatile boolean active;
private final AtomicLong correlationId = new AtomicLong();
private final String gatewayCorrelation = UUID.randomUUID().toString();
private final Map<String, LinkedBlockingQueue<javax.jms.Message>> replies =
new ConcurrentHashMap<String, LinkedBlockingQueue<javax.jms.Message>>();
private final ConcurrentHashMap<String, TimedReply> earlyOrLateReplies =
new ConcurrentHashMap<String, JmsOutboundGateway.TimedReply>();
private volatile ScheduledFuture<?> reaper;
private final Object lifeCycleMonitor = new Object();
private volatile boolean requiresReply;
private long lastSend;
@@ -210,7 +215,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
*/
public void setRequestDestinationExpression(Expression requestDestinationExpression) {
Assert.notNull(requestDestinationExpression, "'requestDestinationExpression' must not be null");
this.requestDestinationExpressionProcessor = new ExpressionEvaluatingMessageProcessor<Object>(requestDestinationExpression);
this.requestDestinationExpressionProcessor =
new ExpressionEvaluatingMessageProcessor<Object>(requestDestinationExpression);
}
/**
@@ -409,7 +415,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
/**
* @param replyContainerProperties the replyContainerproperties to set
* @param replyContainerProperties the replyContainerProperties to set
*/
public void setReplyContainerProperties(ReplyContainerProperties replyContainerProperties) {
this.replyContainerProperties = replyContainerProperties;
@@ -473,7 +479,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
return this.resolveRequestDestination((String) result, session);
}
throw new MessageDeliveryException(message,
"Evaluation of requestDestinationExpression failed to produce a Destination or destination name. Result was: " + result);
"Evaluation of requestDestinationExpression failed " +
"to produce a Destination or destination name. Result was: " + result);
}
throw new MessageDeliveryException(message,
"No requestDestination, requestDestinationName, or requestDestinationExpression has been configured.");
@@ -502,7 +509,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
return this.resolveReplyDestination((String) result, session);
}
throw new MessageDeliveryException(message,
"Evaluation of replyDestinationExpression failed to produce a Destination or destination name. Result was: " + result);
"Evaluation of replyDestinationExpression failed to produce a Destination or destination name. " +
"Result was: " + result);
}
return session.createTemporaryQueue();
}
@@ -524,7 +532,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
Assert.isTrue(this.requestDestination != null
^ this.requestDestinationName != null
^ this.requestDestinationExpressionProcessor != null,
"Exactly one of 'requestDestination', 'requestDestinationName', or 'requestDestinationExpression' is required.");
"Exactly one of 'requestDestination', 'requestDestinationName', " +
"or 'requestDestinationExpression' is required.");
if (this.requestDestinationExpressionProcessor != null) {
this.requestDestinationExpressionProcessor.setBeanFactory(getBeanFactory());
this.requestDestinationExpressionProcessor.setConversionService(getConversionService());
@@ -541,7 +550,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
(this.replyDestination != null || this.replyDestinationName != null) ||
this.replyDestinationExpressionProcessor != null)) {
if (logger.isWarnEnabled()) {
logger.warn("The gateway cannot use a reply listener container with a specified destination(Name/Expression) " +
logger.warn("The gateway cannot use a reply listener container with a specified " +
"destination(Name/Expression) " +
"without a 'correlation-key'; " +
"a container will NOT be used; " +
"to avoid this problem, set the 'correlation-key' attribute; " +
@@ -560,6 +570,16 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
setContainerProperties(container);
container.afterPropertiesSet();
this.replyContainer = container;
if (isAsync() && this.correlationKey == null) {
logger.warn("'async=true' requires a correlationKey; ignored");
setAsync(false);
}
}
else {
if (isAsync()) {
logger.warn("'async=true' is ignored when a reply container is not being used");
setAsync(false);
}
}
this.initialized = true;
}
@@ -611,7 +631,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
container.setRecoveryInterval(this.replyContainerProperties.getRecoveryInterval());
}
if (StringUtils.hasText(this.replyContainerProperties.getSessionAcknowledgeModeName())) {
Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(this.replyContainerProperties.getSessionAcknowledgeModeName());
Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(
this.replyContainerProperties.getSessionAcknowledgeModeName());
if (acknowledgeMode != null) {
if (JmsAdapterUtils.SESSION_TRANSACTED == acknowledgeMode) {
container.setSessionTransacted(true);
@@ -659,7 +680,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
else {
Assert.state(taskScheduler != null, "'taskScheduler' is required.");
}
if (this.receiveTimeout >= 0) {
if (!isAsync() && this.receiveTimeout >= 0) {
Assert.state(taskScheduler != null, "'taskScheduler' is required.");
this.reaper = taskScheduler.schedule(new LateReplyReaper(), new Date());
}
@@ -675,7 +696,9 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
if (this.replyContainer != null) {
this.replyContainer.stop();
this.deleteDestinationIfTemporary(this.replyContainer.getDestination());
this.reaper.cancel(false);
if (this.reaper != null) {
this.reaper.cancel(false);
}
}
if (this.idleTask != null) {
this.idleTask.cancel(true);
@@ -691,15 +714,14 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
@Override
protected Object handleRequestMessage(final Message<?> message) {
protected Object handleRequestMessage(final Message<?> requestMessage) {
if (!this.initialized) {
this.afterPropertiesSet();
afterPropertiesSet();
}
final Message<?> requestMessage = this.getMessageBuilderFactory().fromMessage(message).build();
try {
javax.jms.Message jmsReply;
Object reply;
if (this.replyContainer == null) {
jmsReply = this.sendAndReceiveWithoutContainer(requestMessage);
reply = sendAndReceiveWithoutContainer(requestMessage);
}
else {
if (this.idleReplyContainerTimeout > 0) {
@@ -715,45 +737,53 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
}
}
jmsReply = this.sendAndReceiveWithContainer(requestMessage);
reply = this.sendAndReceiveWithContainer(requestMessage);
}
if (jmsReply == null) {
if (reply == null) {
if (this.requiresReply) {
throw new MessageTimeoutException(message,
throw new MessageTimeoutException(requestMessage,
"failed to receive JMS response within timeout of: " + this.receiveTimeout + "ms");
}
else {
return null;
}
}
Object result = jmsReply;
if (this.extractReplyPayload) {
result = this.messageConverter.fromMessage(jmsReply);
if (logger.isDebugEnabled()) {
logger.debug("converted JMS Message [" + jmsReply + "] to integration Message payload [" + result + "]");
}
}
Map<String, Object> 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 = this.getMessageBuilderFactory().fromMessage((Message<?>) result).copyHeaders(jmsReplyHeaders).build();
if (reply instanceof javax.jms.Message) {
return buildReply((javax.jms.Message) reply);
}
else {
replyMessage = this.getMessageBuilderFactory().withPayload(result).copyHeaders(jmsReplyHeaders).build();
return reply;
}
return replyMessage;
}
catch (JMSException e) {
throw new MessageHandlingException(requestMessage, e);
}
}
private javax.jms.Message sendAndReceiveWithContainer(Message<?> requestMessage) throws JMSException {
private AbstractIntegrationMessageBuilder<?> buildReply(javax.jms.Message jmsReply) throws JMSException {
Object result = jmsReply;
if (this.extractReplyPayload) {
result = this.messageConverter.fromMessage(jmsReply);
if (logger.isDebugEnabled()) {
logger.debug("converted JMS Message [" + jmsReply + "] to integration Message payload [" + result + "]");
}
}
Map<String, Object> 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);
}
if (result instanceof Message){
return getMessageBuilderFactory().fromMessage((Message<?>) result).copyHeaders(jmsReplyHeaders);
}
else {
return getMessageBuilderFactory().withPayload(result).copyHeaders(jmsReplyHeaders);
}
}
private Object sendAndReceiveWithContainer(Message<?> requestMessage) throws JMSException {
Connection connection = this.createConnection();
Session session = null;
Destination replyTo = this.replyContainer.getReplyDestination();
@@ -782,7 +812,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
Destination requestDestination = this.determineRequestDestination(requestMessage, session);
javax.jms.Message reply = null;
Object reply = null;
if (this.correlationKey == null) {
/*
* Remove any existing correlation id that was mapped from the inbound message
@@ -798,8 +828,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
* Remove the gateway's internal correlation Id to avoid conflicts with an upstream
* gateway.
*/
if (reply != null) {
reply.setJMSCorrelationID(null);
if (reply instanceof javax.jms.Message) {
((javax.jms.Message) reply).setJMSCorrelationID(null);
}
return reply;
}
@@ -840,13 +870,16 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
javax.jms.Message replyMessage = null;
Destination requestDestination = this.determineRequestDestination(requestMessage, session);
if (this.correlationKey != null) {
replyMessage = this.doSendAndReceiveWithGeneratedCorrelationId(requestDestination, jmsRequest, replyTo, session, priority);
replyMessage = doSendAndReceiveWithGeneratedCorrelationId(requestDestination, jmsRequest, replyTo,
session, priority);
}
else if (replyTo instanceof TemporaryQueue || replyTo instanceof TemporaryTopic) {
replyMessage = this.doSendAndReceiveWithTemporaryReplyToDestination(requestDestination, jmsRequest, replyTo, session, priority);
replyMessage = doSendAndReceiveWithTemporaryReplyToDestination(requestDestination, jmsRequest, replyTo,
session, priority);
}
else {
replyMessage = this.doSendAndReceiveWithMessageIdCorrelation(requestDestination, jmsRequest, replyTo, session, priority);
replyMessage = doSendAndReceiveWithMessageIdCorrelation(requestDestination, jmsRequest, replyTo,
session, priority);
}
return replyMessage;
}
@@ -858,7 +891,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
/**
* Creates the MessageConsumer before sending the request Message since we are generating our own correlationId value for the MessageSelector.
* Creates the MessageConsumer before sending the request Message since we are generating
* our own correlationId value for the MessageSelector.
*/
private javax.jms.Message doSendAndReceiveWithGeneratedCorrelationId(Destination requestDestination,
javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
@@ -911,7 +945,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
/**
* Creates the MessageConsumer after sending the request Message since we need the MessageID for correlation with a MessageSelector.
* Creates the MessageConsumer after sending the request Message since we need
* the MessageID for correlation with a MessageSelector.
*/
private javax.jms.Message doSendAndReceiveWithMessageIdCorrelation(Destination requestDestination,
javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
@@ -1009,7 +1044,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
}
private javax.jms.Message doSendAndReceiveAsync(Destination requestDestination, javax.jms.Message jmsRequest, Session session, int priority) throws JMSException {
private Object doSendAndReceiveAsync(Destination requestDestination, javax.jms.Message jmsRequest, Session session,
int priority) throws JMSException {
String correlationId = null;
MessageProducer messageProducer = null;
try {
@@ -1026,19 +1062,32 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
*/
jmsRequest.setJMSCorrelationID(null);
}
LinkedBlockingQueue<javax.jms.Message> replyQueue = new LinkedBlockingQueue<javax.jms.Message>(1);
LinkedBlockingQueue<javax.jms.Message> replyQueue = null;
if (logger.isDebugEnabled()) {
logger.debug(this.getComponentName() + " Sending message with correlationId " + correlationId);
}
this.replies.put(correlationId, replyQueue);
SettableListenableFuture<AbstractIntegrationMessageBuilder<?>> future = null;
boolean async = isAsync();
if (!async) {
replyQueue = new LinkedBlockingQueue<javax.jms.Message>(1);
this.replies.put(correlationId, replyQueue);
}
else {
future = createFuture(correlationId);
}
this.sendRequestMessage(jmsRequest, messageProducer, priority);
return obtainReplyFromContainer(correlationId, replyQueue);
if (async) {
return future;
}
else {
return obtainReplyFromContainer(correlationId, replyQueue);
}
}
finally {
JmsUtils.closeMessageProducer(messageProducer);
if (correlationId != null) {
if (correlationId != null && !isAsync()) {
this.replies.remove(correlationId);
}
}
@@ -1103,16 +1152,58 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
if (logger.isDebugEnabled()) {
if (reply == null) {
logger.debug(this.getComponentName() + " Timed out waiting for reply with CorrelationId " + correlationId);
if (logger.isDebugEnabled()) {
logger.debug(this.getComponentName() + " Timed out waiting for reply with CorrelationId "
+ correlationId);
}
}
else {
logger.debug(this.getComponentName() + " Obtained reply with CorrelationId " + correlationId);
if (logger.isDebugEnabled()) {
logger.debug(this.getComponentName() + " Obtained reply with CorrelationId " + correlationId);
}
}
}
return reply;
}
private void sendRequestMessage(javax.jms.Message jmsRequest, MessageProducer messageProducer, int priority) throws JMSException {
private SettableListenableFuture<AbstractIntegrationMessageBuilder<?>> createFuture(final String correlationId) {
SettableListenableFuture<AbstractIntegrationMessageBuilder<?>> future =
new SettableListenableFuture<AbstractIntegrationMessageBuilder<?>>();
this.futures.put(correlationId, future);
if (this.receiveTimeout > 0) {
getTaskScheduler().schedule(new Runnable() {
@Override
public void run() {
expire(correlationId);
}
}, new Date(System.currentTimeMillis() + this.receiveTimeout));
}
return future;
}
private void expire(String correlationId) {
final SettableListenableFuture<AbstractIntegrationMessageBuilder<?>> future = this.futures.remove(correlationId);
if (future != null) {
try {
if (getRequiresReply()) {
future.setException(new JmsTimeoutException("No reply in " + this.receiveTimeout + " ms"));
}
else {
if (logger.isDebugEnabled()) {
logger.debug("Reply expired and reply not required for " + correlationId);
}
}
}
catch (Exception e) {
logger.error("Exception while expiring future");
}
}
}
private void sendRequestMessage(javax.jms.Message jmsRequest, MessageProducer messageProducer, int priority)
throws JMSException {
if (this.explicitQosEnabled) {
messageProducer.send(jmsRequest, this.deliveryMode, priority, this.timeToLive);
}
@@ -1181,6 +1272,33 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
correlationId = message.getStringProperty(this.correlationKey);
}
Assert.state(correlationId != null, "Message with no correlationId received");
if (isAsync()) {
onMessageAsync(message, correlationId);
}
else {
onMessageSync(message, correlationId);
}
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("Failed to consume reply with correlationId " + correlationId, e);
}
}
}
private void onMessageAsync(javax.jms.Message message, String correlationId) throws Exception {
SettableListenableFuture<AbstractIntegrationMessageBuilder<?>> future = this.futures.remove(correlationId);
if (future != null) {
message.setJMSCorrelationID(null);
future.set(buildReply(message));
}
else {
logger.warn("Late reply for " + correlationId);
}
}
private void onMessageSync(javax.jms.Message message, String correlationId) {
try {
LinkedBlockingQueue<javax.jms.Message> queue = this.replies.get(correlationId);
if (queue == null) {
if (this.correlationKey != null) {
@@ -1330,7 +1448,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
if (logger.isTraceEnabled()) {
logger.trace("Running late reply reaper");
}
Iterator<Entry<String, TimedReply>> lateReplyIterator = JmsOutboundGateway.this.earlyOrLateReplies.entrySet().iterator();
Iterator<Entry<String, TimedReply>> lateReplyIterator =
JmsOutboundGateway.this.earlyOrLateReplies.entrySet().iterator();
long now = System.currentTimeMillis();
long expired = now - (JmsOutboundGateway.this.receiveTimeout * 2);
while (lateReplyIterator.hasNext()) {
@@ -1355,7 +1474,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
@Override
public void run() {
synchronized(JmsOutboundGateway.this.lifeCycleMonitor) {
if (System.currentTimeMillis() - JmsOutboundGateway.this.lastSend > JmsOutboundGateway.this.idleReplyContainerTimeout
if (System.currentTimeMillis() - JmsOutboundGateway.this.lastSend >
JmsOutboundGateway.this.idleReplyContainerTimeout
&& JmsOutboundGateway.this.replies.size() == 0) {
if (JmsOutboundGateway.this.replyContainer.isRunning()) {
if (logger.isDebugEnabled()) {

35
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsTimeoutException.java Normal file
View File

@@ -0,0 +1,35 @@
/*
* Copyright 2016 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 org.springframework.jms.JmsException;
/**
* A timeout occurred within an async gateway.
*
* @author Gary Russell
* @since 4.3
*
*/
@SuppressWarnings("serial")
public class JmsTimeoutException extends JmsException {
public JmsTimeoutException(String description) {
super(description);
}
}

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

@@ -69,6 +69,7 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "idle-reply-listener-timeout",
"idleReplyContainerTimeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async");
String deliveryPersistent = element.getAttribute("delivery-persistent");
if (StringUtils.hasText(deliveryPersistent)) {

10
spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.3.xsd
View File

@@ -1128,6 +1128,16 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="async" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
When false (default), the requesting thread is suspended until a reply is received
or a timeout occurs; when true, the requesting thread is released and the reply
is returned on the listener container thread. Requires a reply-listener child
element and a 'correlation-key'; otherwise this property is ignored.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:smartLifeCycleAttributeGroup"/>
</xsd:complexType>
</xsd:element>

7
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsOutboundGatewayParserTests.java
View File

@@ -86,6 +86,7 @@ public class JmsOutboundGatewayParserTests {
accessor = new DirectFieldAccessor(gateway);
int deliveryMode = (Integer)accessor.getPropertyValue("deliveryMode");
assertEquals(DeliveryMode.PERSISTENT, deliveryMode);
assertTrue(TestUtils.getPropertyValue(gateway, "async", Boolean.class));
DefaultMessageListenerContainer container = TestUtils.getPropertyValue(gateway, "replyContainer",
DefaultMessageListenerContainer.class);
assertEquals(4, TestUtils.getPropertyValue(container, "concurrentConsumers"));
@@ -116,6 +117,7 @@ public class JmsOutboundGatewayParserTests {
"jmsOutboundGatewayWithDeliveryPersistent.xml", this.getClass());
EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("advised");
JmsOutboundGateway gateway = TestUtils.getPropertyValue(endpoint, "handler", JmsOutboundGateway.class);
assertFalse(TestUtils.getPropertyValue(gateway, "async", Boolean.class));
gateway.handleMessage(new GenericMessage<String>("foo"));
assertEquals(1, adviceCalled);
assertEquals(3, TestUtils.getPropertyValue(gateway, "replyContainer.sessionAcknowledgeMode"));
@@ -287,13 +289,17 @@ public class JmsOutboundGatewayParserTests {
public interface SampleGateway{
String echo(String value);
}
public static class SampleService{
public String echo(String value){
return value.toUpperCase();
}
}
public static class FooAdvice extends AbstractRequestHandlerAdvice {
@@ -305,4 +311,5 @@ public class JmsOutboundGatewayParserTests {
}
}
}

8
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithDeliveryPersistent.xml
View File

@@ -16,14 +16,14 @@
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
http://www.springframework.org/schema/integration/jms
http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">
<context:property-placeholder properties-ref="props"/>
<util:properties id="props">
<prop key="jmsAcknowledgeModeTransacted">transacted</prop>
<prop key="jmsAcknowledgeModeDupsOk">dups-ok</prop>
</util:properties>
<si:channel id="requestChannel"/>
<jms:outbound-gateway id="jmsGateway"
@@ -31,6 +31,8 @@
request-channel="requestChannel"
delivery-persistent="true"
idle-reply-listener-timeout="1234"
async="true"
correlation-key="JMSCorrelationID"
auto-startup="false">
<jms:reply-listener
acknowledge="${jmsAcknowledgeModeTransacted}"

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

@@ -0,0 +1,174 @@
/*
* Copyright 2016 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.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.integration.jms.JmsTimeoutException;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.support.JmsHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @since 4.3
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class AsyncGatewayTests {
@Autowired
private CachingConnectionFactory ccf;
@Autowired
private JmsOutboundGateway gateway1;
@Autowired
private JmsOutboundGateway gateway2;
@Test
public void testWithReply() throws Exception {
QueueChannel replies = new QueueChannel();
this.gateway1.setOutputChannel(replies);
this.gateway1.start();
this.gateway1.handleMessage(MessageBuilder.withPayload("foo")
.setHeader(JmsHeaders.CORRELATION_ID, "baz")// make sure it's restored in case we're from an upstream gw
.build());
JmsTemplate template = new JmsTemplate(this.ccf);
template.setReceiveTimeout(10000);
final Message received = template.receive("asyncTest1");
assertNotNull(received);
template.send(received.getJMSReplyTo(), new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
TextMessage textMessage = session.createTextMessage("bar");
textMessage.setJMSCorrelationID(received.getJMSCorrelationID());
return textMessage;
}
});
org.springframework.messaging.Message<?> reply = replies.receive(10000);
assertNotNull(reply);
assertEquals("bar", reply.getPayload());
assertEquals("baz", reply.getHeaders().get(JmsHeaders.CORRELATION_ID));
this.gateway1.stop();
}
@Test
public void testWithTimeout() throws Exception {
QueueChannel errors = new QueueChannel();
this.gateway2.setOutputChannel(errors);
this.gateway2.start();
this.gateway2.handleMessage(MessageBuilder.withPayload("foo").setErrorChannel(errors).build());
JmsTemplate template = new JmsTemplate(this.ccf);
final Message received = template.receive("asyncTest3");
assertNotNull(received);
org.springframework.messaging.Message<?> error = errors.receive(10000);
assertNotNull(error);
assertThat(error, instanceOf(ErrorMessage.class));
assertThat(error.getPayload(), instanceOf(MessagingException.class));
assertThat(((MessagingException) error.getPayload()).getCause(), instanceOf(JmsTimeoutException.class));
assertEquals("foo", ((MessagingException) error.getPayload()).getFailedMessage().getPayload());
this.gateway1.stop();
}
@Test
@DirtiesContext
public void testWithTimeoutNoReplyRequired() throws Exception {
QueueChannel errors = new QueueChannel();
this.gateway2.setOutputChannel(errors);
this.gateway2.setRequiresReply(false);
this.gateway2.start();
this.gateway2.handleMessage(MessageBuilder.withPayload("foo").setErrorChannel(errors).build());
JmsTemplate template = new JmsTemplate(this.ccf);
final Message received = template.receive("asyncTest3");
assertNotNull(received);
org.springframework.messaging.Message<?> error = errors.receive(1000);
assertNull(error);
this.gateway1.stop();
}
@Configuration
@EnableIntegration
public static class Config {
@Bean
public CachingConnectionFactory ccf() {
return new CachingConnectionFactory(
new ActiveMQConnectionFactory("vm://localhosti?broker.persistent=false"));
}
@Bean
public JmsOutboundGateway gateway1() {
JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setUseReplyContainer(true);
gateway.setConnectionFactory(ccf());
gateway.setRequestDestinationName("asyncTest1");
gateway.setReplyDestinationName("asyncTest2");
gateway.setRequiresReply(true);
gateway.setReceiveTimeout(10000);
gateway.setAsync(true);
gateway.setCorrelationKey("JMSCorrelationID");
return gateway;
}
@Bean
public JmsOutboundGateway gateway2() {
JmsOutboundGateway gateway = new JmsOutboundGateway();
gateway.setUseReplyContainer(true);
gateway.setConnectionFactory(ccf());
gateway.setRequestDestinationName("asyncTest3");
gateway.setReplyDestinationName("asyncTest4");
gateway.setRequiresReply(true);
gateway.setReceiveTimeout(10);
gateway.setAsync(true);
gateway.setCorrelationKey("JMSCorrelationID");
return gateway;
}
}
}