From 4bfcdb9dfaf270b76f8b30eddc3ae64f2bcd48a0 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 23 Mar 2016 10:12:13 -0400 Subject: [PATCH] INT-3944: Async JMS Outbound Gateway 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()` --- .../outbound/AsyncAmqpOutboundGateway.java | 2 +- ...stractSimpleMessageHandlerFactoryBean.java | 12 +- .../config/xml/ServiceActivatorParser.java | 2 +- .../AbstractMessageProducingHandler.java | 19 +- .../AbstractReplyProducingMessageHandler.java | 4 +- .../routingslip/RoutingSlipRouteStrategy.java | 7 + .../AbstractIntegrationMessageBuilder.java | 4 + .../integration/support/MessageBuilder.java | 19 +- .../support/MutableMessageBuilder.java | 13 +- .../handler/AsyncHandlerTests.java | 2 +- .../LastModifiedFileListFilterTests.java | 2 +- .../integration/jms/JmsOutboundGateway.java | 254 +++++++++++++----- .../integration/jms/JmsTimeoutException.java | 35 +++ .../jms/config/JmsOutboundGatewayParser.java | 1 + .../jms/config/spring-integration-jms-4.3.xsd | 10 + .../config/JmsOutboundGatewayParserTests.java | 7 + ...sOutboundGatewayWithDeliveryPersistent.xml | 8 +- .../jms/request_reply/AsyncGatewayTests.java | 174 ++++++++++++ .../ChatMessageListeningEndpointTests.java | 18 +- src/reference/asciidoc/amqp.adoc | 6 +- src/reference/asciidoc/jms.adoc | 28 +- src/reference/asciidoc/router.adoc | 228 +--------------- src/reference/asciidoc/service-activator.adoc | 2 +- src/reference/asciidoc/whats-new.adoc | 7 + 24 files changed, 539 insertions(+), 325 deletions(-) create mode 100644 spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsTimeoutException.java create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/AsyncGatewayTests.java diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java index 3816b2f7c8..bd4b098348 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AsyncAmqpOutboundGateway.java @@ -49,7 +49,7 @@ public class AsyncAmqpOutboundGateway extends AbstractAmqpOutboundEndpoint { this.messageConverter = template.getMessageConverter(); Assert.notNull(this.messageConverter, "the template's message converter cannot be null"); setConnectionFactory(this.template.getConnectionFactory()); - setAsyncReplySupported(true); + setAsync(true); } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java index 6f461869d9..6526fe5278 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java @@ -84,7 +84,7 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean channelResolver; - private Boolean asyncReplySupported; + private Boolean async; @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { @@ -150,11 +150,11 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean}. - * @param asyncReplySupported the asyncReplySupported to set. + * @param async the async to set. * @since 4.3 */ - public void setAsyncReplySupported(Boolean asyncReplySupported) { - this.asyncReplySupported = asyncReplySupported; + public void setAsync(Boolean async) { + this.async = async; } /** @@ -223,10 +223,10 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean) { + if (this.async && reply instanceof ListenableFuture) { ListenableFuture future = (ListenableFuture) reply; final Object theReplyChannel = replyChannel; future.addCallback(new ListenableFutureCallback() { @@ -230,7 +229,7 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan } else { throw new IllegalArgumentException("The RoutingSlip 'path' can be of " + - "String or RoutingSlipRouteStrategy type, but gotten: " + path); + "String or RoutingSlipRouteStrategy type, but got: " + path.getClass()); } if (routingSlipPathValue instanceof MessageChannel) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 1d46e0c93c..5dbcba0edf 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -107,11 +107,11 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa if (result != null) { sendOutputs(result, message); } - else if (this.requiresReply && !getAsyncReplySupported()) { + else if (this.requiresReply && !isAsync()) { throw new ReplyRequiredException(message, "No reply produced by handler '" + getComponentName() + "', and its 'requiresReply' property is set to true."); } - else if (!getAsyncReplySupported() && logger.isDebugEnabled()) { + else if (!isAsync() && logger.isDebugEnabled()) { logger.debug("handler '" + this + "' produced no reply for request Message: " + message); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/routingslip/RoutingSlipRouteStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/routingslip/RoutingSlipRouteStrategy.java index 95d36e13c9..24ddda5b49 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/routingslip/RoutingSlipRouteStrategy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/routingslip/RoutingSlipRouteStrategy.java @@ -29,6 +29,13 @@ import org.springframework.messaging.Message; */ public interface RoutingSlipRouteStrategy { + /** + * Get the next path for this routing slip. + * @param requestMessage the request message. + * @param reply the reply - depending on context, this may be a user-level domain + * object, a {@link Message} or a {@code AbstractIntegrationMessageBuilder}. + * @return a channel name or another {@link RoutingSlipRouteStrategy}. + */ Object getNextPath(Message requestMessage, Object reply); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java index 5191b81c43..ae5db552c9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/AbstractIntegrationMessageBuilder.java @@ -36,6 +36,10 @@ import org.springframework.util.Assert; */ public abstract class AbstractIntegrationMessageBuilder { + public abstract T getPayload(); + + public abstract Map getHeaders(); + /** * Set the value for the given header name. If the provided value is null, the header will be removed. * diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java index 38ef78c976..933d289e92 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-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. @@ -38,6 +38,7 @@ import org.springframework.util.Assert; * @author Oleg Zhurakousky * @author Dave Syer * @author Gary Russell + * @author Artem Bilan */ public final class MessageBuilder extends AbstractIntegrationMessageBuilder { @@ -62,6 +63,16 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder getHeaders() { + return this.headerAccessor.toMap(); + } + /** * Create a builder for a new {@link Message} instance pre-populated with all of the headers copied from the * provided message. The payload of the provided Message will also be used as the payload for the new message. @@ -72,8 +83,7 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder MessageBuilder fromMessage(Message message) { Assert.notNull(message, "message must not be null"); - MessageBuilder builder = new MessageBuilder(message.getPayload(), message); - return builder; + return new MessageBuilder(message.getPayload(), message); } /** @@ -84,8 +94,7 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder MessageBuilder withPayload(T payload) { - MessageBuilder builder = new MessageBuilder(payload, null); - return builder; + return new MessageBuilder(payload, null); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageBuilder.java index 8d87e13530..3d9889d7a5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2015 the original author or authors. + * Copyright 2014-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. @@ -30,6 +30,7 @@ import org.springframework.util.StringUtils; /** * @author Gary Russell + * @author Artem Bilan * @since 4.0 * */ @@ -53,6 +54,16 @@ public class MutableMessageBuilder extends AbstractIntegrationMessageBuilder< this.headers = this.mutableMessage.getRawHeaders(); } + @Override + public T getPayload() { + return this.mutableMessage.getPayload(); + } + + @Override + public Map getHeaders() { + return this.headers; + } + /** * Create a builder for a new {@link Message} instance pre-populated with all of the headers copied from the * provided message. The payload of the provided Message will also be used as the payload for the new message. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java index 7a0162e2a2..9cfe439dad 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/AsyncHandlerTests.java @@ -111,7 +111,7 @@ public class AsyncHandlerTests { } }; - this.handler.setAsyncReplySupported(true); + this.handler.setAsync(true); this.handler.setOutputChannel(this.output); this.handler.setBeanFactory(mock(BeanFactory.class)); this.latch = new CountDownLatch(1); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/LastModifiedFileListFilterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/LastModifiedFileListFilterTests.java index ccf3683db4..0f4c1b1e7d 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/LastModifiedFileListFilterTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/LastModifiedFileListFilterTests.java @@ -45,7 +45,7 @@ public class LastModifiedFileListFilterTests { fileOutputStream.write("x".getBytes()); fileOutputStream.close(); assertEquals(0, filter.filterFiles(new File[] { foo }).size()); - Thread.sleep(2000); + foo.setLastModified(System.currentTimeMillis() - 10000); assertEquals(1, filter.filterFiles(new File[] { foo }).size()); } 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 10b194c572..d89cea52ae 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 @@ -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> replies = + new ConcurrentHashMap>(); + + private final ConcurrentHashMap earlyOrLateReplies = + new ConcurrentHashMap(); + + private final Map>> futures = + new ConcurrentHashMap>>(); + + 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> replies = - new ConcurrentHashMap>(); - - private final ConcurrentHashMap earlyOrLateReplies = - new ConcurrentHashMap(); - 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(requestDestinationExpression); + this.requestDestinationExpressionProcessor = + new ExpressionEvaluatingMessageProcessor(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 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 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 replyQueue = new LinkedBlockingQueue(1); + LinkedBlockingQueue replyQueue = null; if (logger.isDebugEnabled()) { logger.debug(this.getComponentName() + " Sending message with correlationId " + correlationId); } - this.replies.put(correlationId, replyQueue); + SettableListenableFuture> future = null; + boolean async = isAsync(); + if (!async) { + replyQueue = new LinkedBlockingQueue(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> createFuture(final String correlationId) { + SettableListenableFuture> future = + new SettableListenableFuture>(); + 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> 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> 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 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> lateReplyIterator = JmsOutboundGateway.this.earlyOrLateReplies.entrySet().iterator(); + Iterator> 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()) { diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsTimeoutException.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsTimeoutException.java new file mode 100644 index 0000000000..7e4cdc3b54 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsTimeoutException.java @@ -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); + } + +} 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 b45eeaf3e7..c1e670eef1 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 @@ -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)) { diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.3.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.3.xsd index 4b3eee5a9e..4b0257d409 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.3.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-4.3.xsd @@ -1128,6 +1128,16 @@ + + + + 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. + + + 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 b411b29c23..c9f0b8cd37 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 @@ -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("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 { } } + } 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 ef85b04198..608f1bc1f3 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 @@ -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"> - + - + transacted dups-ok - + 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; + } + + } + +} diff --git a/spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/inbound/ChatMessageListeningEndpointTests.java b/spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/inbound/ChatMessageListeningEndpointTests.java index 3fe3d00511..bf7623e23b 100644 --- a/spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/inbound/ChatMessageListeningEndpointTests.java +++ b/spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/inbound/ChatMessageListeningEndpointTests.java @@ -20,14 +20,18 @@ import static org.hamcrest.core.IsInstanceOf.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.willAnswer; +import static org.mockito.Matchers.anyString; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import java.io.StringReader; import java.util.HashSet; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.apache.commons.logging.Log; import org.jivesoftware.smack.SmackException.NotConnectedException; @@ -198,6 +202,17 @@ public class ChatMessageListeningEndpointTests { Log logger = Mockito.spy(TestUtils.getPropertyValue(endpoint, "logger", Log.class)); given(logger.isInfoEnabled()).willReturn(true); + final CountDownLatch logLatch = new CountDownLatch(1); + willAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + Object result = invocation.callRealMethod(); + logLatch.countDown(); + return result; + } + + }).given(logger).info(anyString()); new DirectFieldAccessor(endpoint).setPropertyValue("logger", logger); @@ -210,8 +225,9 @@ public class ChatMessageListeningEndpointTests { ArgumentCaptor argumentCaptor = new ArgumentCaptor(); - verify(logger).info(argumentCaptor.capture()); + assertTrue(logLatch.await(10, TimeUnit.SECONDS)); + verify(logger).info(argumentCaptor.capture()); assertEquals("The XMPP Message [" + smackMessage + "] with empty body is ignored.", argumentCaptor.getValue()); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index e147ba06ec..c78ae7c27f 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -915,8 +915,10 @@ public class AmqpJavaApplication { The gateway discussed in the previous section is synchronous, in that the sending thread is suspended until a reply is received (or a timeout occurs). Spring Integration _version 4.3_ added this asynchronous gateway, which uses the `AsyncRabbitTemplate` from Spring AMQP. -When a message is sent, the thread returns immediately and the reply is sent on the template's listener container -thread when it is received. +When a message is sent, the thread returns immediately after the send completes, and the reply is sent on the template's +listener container thread when it is received. +This can be useful when the gateway is invoked on a poller thread; the thread is released and is available for other +tasks in the framework. Configuration for an AMQP Async Outbound Gateway is shown below. diff --git a/src/reference/asciidoc/jms.adoc b/src/reference/asciidoc/jms.adoc index 3bb287af3c..cfcfef7152 100644 --- a/src/reference/asciidoc/jms.adoc +++ b/src/reference/asciidoc/jms.adoc @@ -397,6 +397,23 @@ __NOT ALLOWED__ User-supplied correlation ids are not permitted with a reply listener; the gateway will not initialize with this configuration. +[[jms-async-gateway]] +==== Async Gateway + +Starting with _version 4.3_, you can now specify `async="true"` (or `setAsync(true)`) when configuring the outbound +gateway. + +By default, when a request is sent to the gateway, the requesting thread is suspended until the reply is received and +the flow then continues on that thread. +If `async` is true, the requesting thread is released immediately after the send completes, and the reply is returned +(and the flow continues) on the listener container thread. +This can be useful when the gateway is invoked on a poller thread; the thread is released and is available for other +tasks within the framework. + +`async` requires a `` (or `setUseReplyContainer(true)` when using Java configuration); it also +requires a `correlationKey` (usually `JMSCorrelationID`) to be specified. +If either of these conditions are not met, `async` is ignored. + [[jms-og-attributes]] ==== Attribute Reference @@ -426,9 +443,10 @@ configuration. request-destination-name="" <21> request-pub-sub-domain="" <22> time-to-live="" <23> - requires-reply=""> <24> - idle-reply-listener-timeout <25> - <26> + requires-reply="" <24> + idle-reply-listener-timeout="" <25> + async=""> <26> + <27> ---- @@ -541,7 +559,9 @@ are outstanding). The container will be started again on the next request. The stop time is a minimum and may actually be up to 1.5x this value. -<26> When this element is included, replies are received by an asynchronous `MessageListenerContainer` rather than +<26> See <>. + +<27> When this element is included, replies are received by an asynchronous `MessageListenerContainer` rather than creating a consumer for each reply. This can be more efficient in many cases. diff --git a/src/reference/asciidoc/router.adoc b/src/reference/asciidoc/router.adoc index 321825f5b8..62eb9581b5 100644 --- a/src/reference/asciidoc/router.adoc +++ b/src/reference/asciidoc/router.adoc @@ -28,16 +28,6 @@ In order to provide a quick overview, all available attributes are listed in the [cols="2,1,1,1,1,1,1", options="header"] |=== - - - - - - - - - - | Attribute @@ -59,12 +49,6 @@ In order to provide a quick overview, all available attributes are listed in the | exception type router - - - - - - | apply-sequence @@ -76,12 +60,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | default-output-channel @@ -93,12 +71,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | resolution-required @@ -110,12 +82,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | ignore-send-failures @@ -127,12 +93,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | timeout @@ -144,12 +104,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | id @@ -161,12 +115,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | auto-startup @@ -178,12 +126,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | input-channel @@ -195,12 +137,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | order @@ -212,12 +148,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | method @@ -229,12 +159,6 @@ a| image::images/tickmark.png[] | - - - - - - | ref @@ -246,12 +170,6 @@ a| image::images/tickmark.png[] | - - - - - - | expression @@ -263,12 +181,6 @@ a| image::images/tickmark.png[] | - - - - - - | header-name @@ -280,12 +192,6 @@ a| image::images/tickmark.png[] | - - - - - - | evaluate-as-string @@ -297,12 +203,6 @@ a| image::images/tickmark.png[] | - - - - - - | xpath-expression-ref @@ -314,12 +214,6 @@ a| image::images/tickmark.png[] | - - - - - - | converter @@ -331,10 +225,6 @@ a| image::images/tickmark.png[] | - - - - |=== .Routers Inside of a Chain @@ -342,15 +232,6 @@ a| image::images/tickmark.png[] |=== - - - - - - - - - | Attribute @@ -372,12 +253,6 @@ a| image::images/tickmark.png[] | exception type router - - - - - - | apply-sequence @@ -389,12 +264,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | default-output-channel @@ -406,12 +275,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | resolution-required @@ -423,12 +286,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | ignore-send-failures @@ -440,12 +297,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | timeout @@ -457,12 +308,6 @@ a| image::images/tickmark.png[] a| image::images/tickmark.png[] - - - - - - | id @@ -474,12 +319,6 @@ a| image::images/tickmark.png[] | - - - - - - | auto-startup @@ -491,12 +330,6 @@ a| image::images/tickmark.png[] | - - - - - - | input-channel @@ -508,12 +341,6 @@ a| image::images/tickmark.png[] | - - - - - - | order @@ -525,12 +352,6 @@ a| image::images/tickmark.png[] | - - - - - - | method @@ -542,12 +363,6 @@ a| image::images/tickmark.png[] | - - - - - - | ref @@ -559,12 +374,6 @@ a| image::images/tickmark.png[] | - - - - - - | expression @@ -576,12 +385,6 @@ a| image::images/tickmark.png[] | - - - - - - | header-name @@ -593,12 +396,6 @@ a| image::images/tickmark.png[] | - - - - - - | evaluate-as-string @@ -610,12 +407,6 @@ a| image::images/tickmark.png[] | - - - - - - | xpath-expression-ref @@ -627,12 +418,6 @@ a| image::images/tickmark.png[] | - - - - - - | converter @@ -644,10 +429,6 @@ a| image::images/tickmark.png[] | - - - - |=== [IMPORTANT] @@ -1310,6 +1091,15 @@ This is to avoid the overhead of `EvaluationContext` creation for each `Expressi It is a simple Java Bean with two properties - `Message request` and `Object reply`. With this expression implementation, we can specify Routing Slip `path` entries using SpEL (`@routingSlipRoutingPojo.get(request, reply)`, `request.headers[myRoutingSlipChannel]`) avoiding a bean definition for the `RoutingSlipRouteStrategy`. +NOTE: The `requestMessage` argument is always a `Message`; depending on context, the reply object may be a +`Message`, an `AbstractIntegrationMessageBuilder` or an arbitrary application domain object (if, for example, +it is returned by a POJO method invoked by a service activator). +In the first two cases, the usual "message" properties are available (`payload` and `headers`) when using SpEL (or +a Java implementation). +When an arbitrary domain object, these properties are, obviously, not available. +Care should be taken when using routing slips in conjunction with POJO methods if the result is used to determine the +next path. + IMPORTANT: If a _Routing Slip_ is involved in a distributed environment - cross-JVM application, `request-reply` through a Message Broker (e.g. <>, <>), or persistence `MessageStore` (<>) is used in the integration flow, etc., - it is recommended to *not* use _inline_ expressions for the Routing Slip `path`. The framework (`RoutingSlipHeaderValueMessageProcessor`) converts them to `ExpressionEvaluatingRoutingSlipRouteStrategy` objects and they are used in the `routingSlip` message header. diff --git a/src/reference/asciidoc/service-activator.adoc b/src/reference/asciidoc/service-activator.adoc index 53d8e2d98e..9d3a538832 100644 --- a/src/reference/asciidoc/service-activator.adoc +++ b/src/reference/asciidoc/service-activator.adoc @@ -114,7 +114,7 @@ The service activator is invoked by the calling thread; this would be some upstr `SubscribableChannel`, or a poller thread for a `PollableChannel`. If the service returns a `ListenableFuture` the default action is to send that as the payload of the message sent to the output (or reply) channel. -Starting with _version 4.3_, you can now set the `async` attribute to true (`setAsyncReplySupported(true)` when using +Starting with _version 4.3_, you can now set the `async` attribute to true (`setAsync(true)` when using Java configuration). If the service returns a `ListenableFuture` when this is true, the calling thread is released immediately, and the reply message is sent on the thread (from within your service) that completes the future. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 466197681e..413a882767 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -48,10 +48,17 @@ See <> for more information. ==== JMS Changes +===== Header Mapper + The `DefaultJmsHeaderMapper` now maps the standard `correlationId` header as a message property by invoking its `toString()` method. See <> for more information. +===== Async Gateway + +The JMS Outbound gateway now has an `async` property. +See <> for more information. + ==== Aggregator Changes There is a change in behavior when a POJO aggregator releases a collection of `Message` objects; this is rare but if