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

destination expr for outbound-channel-adapter

Browse Source
This commit is contained in:
Mark Fisher
2011-08-30 17:59:05 -04:00
parent f4cb32c866
commit d3f5c90348
6 changed files with 214 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsSendingMessageHandler.java
View File

@@ -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<Object>(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 {
}
}
}

8
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsAdapterParserUtils.java
View File

@@ -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";

26
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundChannelAdapterParser.java
View File

@@ -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)) {

32
spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.1.xsd
View File

@@ -870,7 +870,8 @@
<xsd:documentation>
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'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -879,8 +880,33 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="destination-name" type="xsd:string"/>
<xsd:attribute name="pub-sub-domain" type="xsd:string"/>
<xsd:attribute name="destination-name" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="destination-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
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'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="pub-sub-domain" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
If true, specifies that destination names should resolve to Topics rather than Queues. Default is false.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>

36
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests-context.xml Normal file
View File

@@ -0,0 +1,36 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">
<int-jms:outbound-channel-adapter id="channelAdapterChannel" destination-expression="'queue.test.dynamic.adapter.' + headers.destinationNumber"/>
<int-jms:message-driven-channel-adapter channel="channelAdapterResults1" destination-name="queue.test.dynamic.adapter.1" extract-payload="false"/>
<int-jms:message-driven-channel-adapter channel="channelAdapterResults2" destination-name="queue.test.dynamic.adapter.2" extract-payload="false"/>
<int:channel id="channelAdapterResults1">
<int:queue capacity="1"/>
</int:channel>
<int:channel id="channelAdapterResults2">
<int:queue capacity="1"/>
</int:channel>
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost"/>
</bean>
</property>
<property name="sessionCacheSize" value="10"/>
<property name="cacheProducers" value="false"/>
</bean>
</beans>

77
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests.java Normal file
View File

@@ -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());
}
}