From d3f5c90348e267bc7b47d21cf3f1d4ee65a41bfa Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 30 Aug 2011 17:59:05 -0400 Subject: [PATCH] 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()); + } + +}