From 49ee745e6fe1c277f3c36138052b467ea3290bf0 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 30 Aug 2011 19:21:37 -0400 Subject: [PATCH] request destination exprs for outbound-gateway --- .../integration/jms/JmsOutboundGateway.java | 75 +++++++++++++++---- .../jms/config/JmsOutboundGatewayParser.java | 20 +++-- .../jms/config/spring-integration-jms-2.1.xsd | 34 ++++++++- .../JmsDynamicDestinationTests-context.xml | 11 +++ .../config/JmsDynamicDestinationTests.java | 29 +++++++ 5 files changed, 145 insertions(+), 24 deletions(-) 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 c10fa186a9..706647822b 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 @@ -33,11 +33,14 @@ import javax.jms.TemporaryQueue; import javax.jms.TemporaryTopic; import javax.jms.Topic; +import org.springframework.expression.Expression; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; +import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor; import org.springframework.integration.support.MessageBuilder; import org.springframework.jms.connection.ConnectionFactoryUtils; import org.springframework.jms.support.JmsUtils; @@ -61,6 +64,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { private volatile String requestDestinationName; + private volatile ExpressionEvaluatingMessageProcessor requestDestinationExpressionProcessor; + private volatile Destination replyDestination; private volatile String replyDestinationName; @@ -121,7 +126,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { /** * Set the JMS Destination to which request Messages should be sent. - * Either this or the 'requestDestinationName' property is required. + * Either this or one of 'requestDestinationName' or 'requestDestinationExpression' is required. */ public void setRequestDestination(Destination requestDestination) { if (requestDestination instanceof Topic) { @@ -131,13 +136,22 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { } /** - * Set the name of the JMS Destination to which request Messages should be - * sent. Either this or the 'requestDestination' property is required. + * Set the name of the JMS Destination to which request Messages should be sent. + * Either this or one of 'requestDestination' or 'requestDestinationExpression' is required. */ public void setRequestDestinationName(String requestDestinationName) { this.requestDestinationName = requestDestinationName; } + /** + * Set the SpEL Expression to be used for determining the request Destination instance + * or request destination name. Either this or one of 'requestDestination' or + * 'requestDestinationName' is required. + */ + public void setRequestDestinationExpression(Expression requestDestinationExpression) { + this.requestDestinationExpressionProcessor = new ExpressionEvaluatingMessageProcessor(requestDestinationExpression); + } + /** * Set the JMS Destination from which reply Messages should be received. * If none is provided, this gateway will create a {@link TemporaryQueue} per invocation. @@ -304,14 +318,33 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { return "jms:outbound-gateway"; } - private Destination getRequestDestination(Session session) throws JMSException { + private Destination getRequestDestination(Message message, Session session) throws JMSException { if (this.requestDestination != null) { return this.requestDestination; } + if (this.requestDestinationName != null) { + return this.resolveRequestDestination(this.requestDestinationName, session); + } + if (this.requestDestinationExpressionProcessor != null) { + Object result = this.requestDestinationExpressionProcessor.processMessage(message); + if (result instanceof Destination) { + return (Destination) result; + } + if (result instanceof String) { + return this.resolveRequestDestination((String) result, session); + } + throw new MessageDeliveryException(message, + "Evaluation of requestDestinationExpression failed to produce a Destination or destination name. Result was: " + result); + } + throw new MessageDeliveryException(message, + "No requestDestination, requestDestinationName, or requestDestinationExpression has been configured."); + } + + private Destination resolveRequestDestination(String requestDestinationName, Session session) throws JMSException { Assert.notNull(this.destinationResolver, "DestinationResolver is required when relying upon the 'requestDestinationName' property."); return this.destinationResolver.resolveDestinationName( - session, this.requestDestinationName, this.requestPubSubDomain); + session, requestDestinationName, this.requestPubSubDomain); } private Destination getReplyDestination(Session session) throws JMSException { @@ -334,8 +367,14 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { return; } Assert.notNull(this.connectionFactory, "connectionFactory must not be null"); - Assert.isTrue(this.requestDestination != null || this.requestDestinationName != null, - "Either a 'requestDestination' or 'requestDestinationName' is required."); + Assert.isTrue(this.requestDestination != null + ^ this.requestDestinationName != null + ^ this.requestDestinationExpressionProcessor != null, + "Exactly one of 'requestDestination', 'requestDestinationName', or 'requestDestinationExpression' is required."); + if (this.requestDestinationExpressionProcessor != null) { + this.requestDestinationExpressionProcessor.setBeanFactory(getBeanFactory()); + this.requestDestinationExpressionProcessor.setConversionService(getConversionService()); + } this.initialized = true; } } @@ -402,14 +441,15 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { priority = this.priority; } javax.jms.Message replyMessage = null; + Destination requestDestination = this.getRequestDestination(requestMessage, session); if (this.correlationKey != null) { - replyMessage = this.doSendAndReceiveWithGeneratedCorrelationId(jmsRequest, replyTo, session, priority); + replyMessage = this.doSendAndReceiveWithGeneratedCorrelationId(requestDestination, jmsRequest, replyTo, session, priority); } else if (replyTo instanceof TemporaryQueue || replyTo instanceof TemporaryTopic) { - replyMessage = this.doSendAndReceiveWithTemporaryReplyToDestination(jmsRequest, replyTo, session, priority); + replyMessage = this.doSendAndReceiveWithTemporaryReplyToDestination(requestDestination, jmsRequest, replyTo, session, priority); } else { - replyMessage = this.doSendAndReceiveWithMessageIdCorrelation(jmsRequest, replyTo, session, priority); + replyMessage = this.doSendAndReceiveWithMessageIdCorrelation(requestDestination, jmsRequest, replyTo, session, priority); } return replyMessage; } @@ -423,11 +463,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { /** * Creates the MessageConsumer before sending the request Message since we are generating our own correlationId value for the MessageSelector. */ - private javax.jms.Message doSendAndReceiveWithGeneratedCorrelationId(javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException { + private javax.jms.Message doSendAndReceiveWithGeneratedCorrelationId(Destination requestDestination, + javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException { MessageProducer messageProducer = null; MessageConsumer messageConsumer = null; try { - messageProducer = session.createProducer(this.getRequestDestination(session)); + messageProducer = session.createProducer(requestDestination); String correlationId = UUID.randomUUID().toString().replaceAll("'", "''"); Assert.state(this.correlationKey != null, "correlationKey must not be null"); String messageSelector = null; @@ -452,11 +493,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { /** * Creates the MessageConsumer before sending the request Message since we do not need any correlation. */ - private javax.jms.Message doSendAndReceiveWithTemporaryReplyToDestination(javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException { + private javax.jms.Message doSendAndReceiveWithTemporaryReplyToDestination(Destination requestDestination, + javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException { MessageProducer messageProducer = null; MessageConsumer messageConsumer = null; try { - messageProducer = session.createProducer(this.getRequestDestination(session)); + messageProducer = session.createProducer(requestDestination); messageConsumer = session.createConsumer(replyTo); this.sendRequestMessage(jmsRequest, messageProducer, priority); return this.receiveReplyMessage(messageConsumer); @@ -470,7 +512,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { /** * Creates the MessageConsumer after sending the request Message since we need the MessageID for correlation with a MessageSelector. */ - private javax.jms.Message doSendAndReceiveWithMessageIdCorrelation(javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException { + private javax.jms.Message doSendAndReceiveWithMessageIdCorrelation(Destination requestDestination, + javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException { if (replyTo instanceof Topic && logger.isWarnEnabled()) { logger.warn("Relying on the MessageID for correlation is not recommended when using a Topic as the replyTo Destination " + "because that ID can only be provided to a MessageSelector after the reuqest Message has been sent thereby " + @@ -481,7 +524,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler { MessageProducer messageProducer = null; MessageConsumer messageConsumer = null; try { - messageProducer = session.createProducer(this.getRequestDestination(session)); + messageProducer = session.createProducer(requestDestination); this.sendRequestMessage(jmsRequest, messageProducer, priority); String messageId = jmsRequest.getJMSMessageID().replaceAll("'", "''"); String messageSelector = "JMSCorrelationID = '" + messageId + "'"; 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 4206c4269b..5f455084e2 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 @@ -20,6 +20,7 @@ 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.util.StringUtils; @@ -44,16 +45,25 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser { builder.addPropertyReference("connectionFactory", element.getAttribute("connection-factory")); String requestDestination = element.getAttribute("request-destination"); String requestDestinationName = element.getAttribute("request-destination-name"); - if (!(StringUtils.hasText(requestDestination) ^ StringUtils.hasText(requestDestinationName))) { - parserContext.getReaderContext().error( - "Exactly one of the 'request-destination' or 'request-destination-name' attributes is required.", element); + String requestDestinationExpression = element.getAttribute("request-destination-expression"); + boolean hasRequestDestination = StringUtils.hasText(requestDestination); + boolean hasRequestDestinationName = StringUtils.hasText(requestDestinationName); + boolean hasRequestDestinationExpression = StringUtils.hasText(requestDestinationExpression); + if (!(hasRequestDestination ^ hasRequestDestinationName ^ hasRequestDestinationExpression)) { + parserContext.getReaderContext().error("Exactly one of the 'request-destination', " + + "'request-destination-name', or 'request-destination-expression' attributes is required.", element); } - if (StringUtils.hasText(requestDestination)) { + if (hasRequestDestination) { builder.addPropertyReference("requestDestination", requestDestination); } - else if (StringUtils.hasText(requestDestinationName)) { + else if (hasRequestDestinationName) { builder.addPropertyValue("requestDestinationName", requestDestinationName); } + else if (hasRequestDestinationExpression) { + BeanDefinitionBuilder expressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class); + expressionBuilder.addConstructorArgValue(requestDestinationExpression); + builder.addPropertyValue("requestDestinationExpression", expressionBuilder.getBeanDefinition()); + } IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-destination"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-destination-name"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel"); diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.1.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.1.xsd index b1cd77e08c..18a6800906 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.1.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.1.xsd @@ -732,7 +732,9 @@ A reference to a javax.jms.Destination by bean name. As an alternative to a bean reference, use 'request-destination-name' and 'request-pub-sub-domain' which will rely - upon the DestinationResolver strategy (DynamicDestinationResolver by default). + upon the DestinationResolver strategy (DynamicDestinationResolver by default). This + attribute is mutually exclusive with 'request-destination-name' and + 'request-destination-expression'. @@ -741,8 +743,34 @@ - - + + + + Name of a destination to which request messages will be sent. This name will be handled + by this gateway's DestinationResolver. This attribute is mutually exclusive with + 'request-destination' and 'request-destination-expression'. + + + + + + + A SpEL expression to be evaluated at runtime against each Spring Integration request Message as + the root object. The result should be either a Destination instance or a String representing + the destination name. In the latter case, it will be passed to this adapter's DestinationResolver. + This attribute is mutually exclusive with 'request-destination' and 'request-destination-name'. + + + + + + + When resolving a request destination name (rather than having a 'request-destination' reference), + a true value here specifies that the DestinationResolver should resolve Topics rather than Queues. + Default is false. + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests-context.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests-context.xml index 260e5edbb2..9cee5d0f67 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests-context.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests-context.xml @@ -15,6 +15,10 @@ + + + + @@ -33,4 +37,11 @@ + + + + + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests.java index 984fc18554..1896ee3802 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests.java @@ -28,6 +28,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; +import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.support.MessageBuilder; import org.springframework.test.context.ContextConfiguration; @@ -44,6 +45,9 @@ public class JmsDynamicDestinationTests { @Autowired private MessageChannel channelAdapterChannel; + @Autowired + private MessageChannel gatewayChannel; + @Autowired private PollableChannel channelAdapterResults1; @@ -74,4 +78,29 @@ public class JmsDynamicDestinationTests { assertEquals("queue://queue.test.dynamic.adapter.2", jmsResult2.getJMSDestination().toString()); } + @Test + public void gateway() throws Exception { + Message message1 = MessageBuilder.withPayload("test-1").setHeader("destinationNumber", 1).build(); + Message message2 = MessageBuilder.withPayload("test-2").setHeader("destinationNumber", 2).build(); + MessagingTemplate template = new MessagingTemplate(); + Message result1 = template.sendAndReceive(gatewayChannel, message1); + Message result2 = template.sendAndReceive(gatewayChannel, message2); + assertNotNull(result1); + assertNotNull(result2); + assertEquals("test-1!", result1.getPayload()); + assertEquals("test-2!!", result2.getPayload()); + } + + + public static class Responder { + + public String one(String message) { + return message + "!"; + } + + public String two(String message) { + return message + "!!"; + } + } + }