From d3f5c90348e267bc7b47d21cf3f1d4ee65a41bfa Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 30 Aug 2011 17:59:05 -0400 Subject: [PATCH 1/2] destination expr for outbound-channel-adapter --- .../jms/JmsSendingMessageHandler.java | 57 +++++++++++--- .../jms/config/JmsAdapterParserUtils.java | 8 +- .../JmsOutboundChannelAdapterParser.java | 26 +++++-- .../jms/config/spring-integration-jms-2.1.xsd | 32 +++++++- .../JmsDynamicDestinationTests-context.xml | 36 +++++++++ .../config/JmsDynamicDestinationTests.java | 77 +++++++++++++++++++ 6 files changed, 214 insertions(+), 22 deletions(-) create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests-context.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests.java diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java index d7bed290f1..2eb202eeb4 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2011 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,8 +19,11 @@ package org.springframework.integration.jms; import javax.jms.Destination; import javax.jms.JMSException; +import org.springframework.expression.Expression; import org.springframework.integration.Message; +import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessagePostProcessor; import org.springframework.util.Assert; @@ -43,21 +46,31 @@ public class JmsSendingMessageHandler extends AbstractMessageHandler { private volatile boolean extractPayload = true; + private volatile ExpressionEvaluatingMessageProcessor destinationExpressionProcessor; + public JmsSendingMessageHandler(JmsTemplate jmsTemplate) { this.jmsTemplate = jmsTemplate; } public void setDestination(Destination destination) { - Assert.isNull(this.destinationName, "The 'destination' and 'destinationName' properties are mutually exclusive."); + Assert.isTrue(this.destinationName == null && this.destinationExpressionProcessor == null, + "The 'destination', 'destinationName', and 'destinationExpression' properties are mutually exclusive."); this.destination = destination; } public void setDestinationName(String destinationName) { - Assert.isNull(this.destination, "The 'destination' and 'destinationName' properties are mutually exclusive."); + Assert.isTrue(this.destination == null && this.destinationExpressionProcessor == null, + "The 'destination', 'destinationName', and 'destinationExpression' properties are mutually exclusive."); this.destinationName = destinationName; } + public void setDestinationExpression(Expression destinationExpression) { + Assert.isTrue(this.destination == null && this.destinationName == null, + "The 'destination', 'destinationName', and 'destinationExpression' properties are mutually exclusive."); + this.destinationExpressionProcessor = new ExpressionEvaluatingMessageProcessor(destinationExpression); + } + public void setHeaderMapper(JmsHeaderMapper headerMapper) { this.headerMapper = headerMapper; } @@ -78,28 +91,55 @@ public class JmsSendingMessageHandler extends AbstractMessageHandler { return "jms:outbound-channel-adapter"; } + @Override + protected void onInit() { + if (this.destinationExpressionProcessor != null) { + this.destinationExpressionProcessor.setBeanFactory(getBeanFactory()); + this.destinationExpressionProcessor.setConversionService(getConversionService()); + } + } + @Override protected void handleMessageInternal(final Message message) throws Exception { if (message == null) { throw new IllegalArgumentException("message must not be null"); } + Object destination = this.determineDestination(message); Object objectToSend = (this.extractPayload) ? message.getPayload() : message; MessagePostProcessor messagePostProcessor = new HeaderMappingMessagePostProcessor(message, this.headerMapper); try { DynamicJmsTemplateProperties.setPriority(message.getHeaders().getPriority()); - this.send(objectToSend, messagePostProcessor); + this.send(destination, objectToSend, messagePostProcessor); } finally { DynamicJmsTemplateProperties.clearPriority(); } } - private void send(Object objectToSend, MessagePostProcessor messagePostProcessor) { + private Object determineDestination(Message message) { if (this.destination != null) { - this.jmsTemplate.convertAndSend(this.destination, objectToSend, messagePostProcessor); + return this.destination; } - else if (this.destinationName != null) { - this.jmsTemplate.convertAndSend(this.destinationName, objectToSend, messagePostProcessor); + if (this.destinationName != null) { + return this.destinationName; + } + if (this.destinationExpressionProcessor != null) { + Object result = this.destinationExpressionProcessor.processMessage(message); + if (!(result instanceof Destination || result instanceof String)) { + throw new MessageDeliveryException(message, + "Evaluation of destinationExpression failed to produce a Destination or destination name. Result was: " + result); + } + return result; + } + return null; + } + + private void send(Object destination, Object objectToSend, MessagePostProcessor messagePostProcessor) { + if (destination instanceof Destination) { + this.jmsTemplate.convertAndSend((Destination) destination, objectToSend, messagePostProcessor); + } + else if (destination instanceof String) { + this.jmsTemplate.convertAndSend((String) destination, objectToSend, messagePostProcessor); } else { // fallback to default destination of the template this.jmsTemplate.convertAndSend(objectToSend, messagePostProcessor); @@ -124,5 +164,4 @@ public class JmsSendingMessageHandler extends AbstractMessageHandler { } } - } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java index 9e1e6a00c7..9b2a5e3a76 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java @@ -46,12 +46,16 @@ abstract class JmsAdapterParserUtils { static final String DESTINATION_NAME_ATTRIBUTE = "destination-name"; + static final String DESTINATION_NAME_PROPERTY = "destinationName"; + + static final String DESTINATION_EXPRESSION_ATTRIBUTE = "destination-expression"; + + static final String DESTINATION_EXPRESSION_PROPERTY = "destinationExpression"; + static final String PUB_SUB_DOMAIN_ATTRIBUTE = "pub-sub-domain"; static final String PUB_SUB_DOMAIN_PROPERTY = "pubSubDomain"; - static final String DESTINATION_NAME_PROPERTY = "destinationName"; - static final String HEADER_MAPPER_ATTRIBUTE = "header-mapper"; static final String HEADER_MAPPER_PROPERTY = "headerMapper"; diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java index 8e2241c076..e55f7196a6 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2011 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ import org.w3c.dom.Element; import org.springframework.beans.factory.support.AbstractBeanDefinition; 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.AbstractOutboundChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.util.StringUtils; @@ -39,10 +40,12 @@ public class JmsOutboundChannelAdapterParser extends AbstractOutboundChannelAdap String jmsTemplate = element.getAttribute(JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE); String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE); String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE); + String destinationExpression = element.getAttribute(JmsAdapterParserUtils.DESTINATION_EXPRESSION_ATTRIBUTE); String headerMapper = element.getAttribute(JmsAdapterParserUtils.HEADER_MAPPER_ATTRIBUTE); boolean hasJmsTemplate = StringUtils.hasText(jmsTemplate); boolean hasDestinationRef = StringUtils.hasText(destination); boolean hasDestinationName = StringUtils.hasText(destinationName); + boolean hasDestinationExpression = StringUtils.hasText(destinationExpression); if (hasJmsTemplate) { JmsAdapterParserUtils.verifyNoJmsTemplateAttributes(element, parserContext); builder.addConstructorArgReference(jmsTemplate); @@ -50,22 +53,29 @@ public class JmsOutboundChannelAdapterParser extends AbstractOutboundChannelAdap else { builder.addConstructorArgValue(JmsAdapterParserUtils.parseJmsTemplateBeanDefinition(element, parserContext)); } - if (hasDestinationRef || hasDestinationName) { + + if (hasDestinationRef || hasDestinationName || hasDestinationExpression) { + if (!(hasDestinationRef ^ hasDestinationName ^ hasDestinationExpression)) { + parserContext.getReaderContext().error("The 'destination', 'destination-name', and " + + "'destination-expression' attributes are mutually exclusive.", parserContext.extractSource(element)); + } if (hasDestinationRef) { - if (hasDestinationName) { - parserContext.getReaderContext().error("The 'destination-name' " + - "and 'destination' attributes are mutually exclusive.", parserContext.extractSource(element)); - } builder.addPropertyReference(JmsAdapterParserUtils.DESTINATION_PROPERTY, destination); } else if (hasDestinationName) { builder.addPropertyValue(JmsAdapterParserUtils.DESTINATION_NAME_PROPERTY, destinationName); } + else if (hasDestinationExpression) { + BeanDefinitionBuilder expressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class); + expressionBuilder.addConstructorArgValue(destinationExpression); + builder.addPropertyValue(JmsAdapterParserUtils.DESTINATION_EXPRESSION_PROPERTY, expressionBuilder.getBeanDefinition()); + } } else if (!hasJmsTemplate) { parserContext.getReaderContext().error("either a '" + JmsAdapterParserUtils.JMS_TEMPLATE_ATTRIBUTE + - "' or one of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "' or '" - + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + + "' or one of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE + "', '" + + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "', or '" + + JmsAdapterParserUtils.DESTINATION_EXPRESSION_ATTRIBUTE + "' attributes must be provided", parserContext.extractSource(element)); } if (StringUtils.hasText(headerMapper)) { 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 06acb59725..b1cd77e08c 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 @@ -870,7 +870,8 @@ A reference to a javax.jms.Destination by bean name. As an alternative to a bean reference, use 'destination-name' and 'pub-sub-domain' which will rely upon the - DestinationResolver strategy (DynamicDestinationResolver by default). + DestinationResolver strategy (DynamicDestinationResolver by default). This attribute + is mutually exclusive with 'destination-name' and 'destination-expression'. @@ -879,8 +880,33 @@ - - + + + + Name of the destination to which JMS Messages will be sent. This will be passed to the + adapter's DestinationResolver. This attribute is mutually exclusive with 'destination' + and 'destination-expression'. + + + + + + + A SpEL expression to be evaluated at runtime against each Spring Integration 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. + If the evaluation result is null, messages will be sent to the default destination of the + underlying JmsTemplate. This attribute is mutually exclusive with 'destination' and 'destination-name'. + + + + + + + If true, specifies that destination names should resolve to 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 new file mode 100644 index 0000000000..260e5edbb2 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests-context.xml @@ -0,0 +1,36 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + 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 new file mode 100644 index 0000000000..984fc18554 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests.java @@ -0,0 +1,77 @@ +/* + * Copyright 2002-2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * 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.config; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import javax.jms.TextMessage; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Mark Fisher + * @since 2.1 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class JmsDynamicDestinationTests { + + @Autowired + private MessageChannel channelAdapterChannel; + + @Autowired + private PollableChannel channelAdapterResults1; + + @Autowired + private PollableChannel channelAdapterResults2; + + + @Before + public void prepareActiveMq() { + ActiveMqTestUtils.prepare(); + } + + @Test + public void channelAdapter() throws Exception { + Message message1 = MessageBuilder.withPayload("test-1").setHeader("destinationNumber", 1).build(); + Message message2 = MessageBuilder.withPayload("test-2").setHeader("destinationNumber", 2).build(); + channelAdapterChannel.send(message1); + channelAdapterChannel.send(message2); + Message result1 = channelAdapterResults1.receive(5000); + Message result2 = channelAdapterResults2.receive(5000); + assertNotNull(result1); + assertNotNull(result2); + TextMessage jmsResult1 = (TextMessage) result1.getPayload(); + TextMessage jmsResult2 = (TextMessage) result2.getPayload(); + assertEquals("test-1", jmsResult1.getText()); + assertEquals("queue://queue.test.dynamic.adapter.1", jmsResult1.getJMSDestination().toString()); + assertEquals("test-2", jmsResult2.getText()); + assertEquals("queue://queue.test.dynamic.adapter.2", jmsResult2.getJMSDestination().toString()); + } + +} From 49ee745e6fe1c277f3c36138052b467ea3290bf0 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 30 Aug 2011 19:21:37 -0400 Subject: [PATCH 2/2] 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 + "!!"; + } + } + }