INT-809 Added namespace support for MessagePublishigInterceptor
This commit is contained in:
Oleg Zhurakousky
@@ -56,6 +56,7 @@ public class IntegrationNamespaceHandler extends AbstractIntegrationNamespaceHan
|
||||
registerBeanDefinitionParser("annotation-config", new AnnotationConfigParser());
|
||||
registerBeanDefinitionParser("application-event-multicaster", new ApplicationEventMulticasterParser());
|
||||
registerBeanDefinitionParser("thread-pool-task-executor", new ThreadPoolTaskExecutorParser());
|
||||
registerBeanDefinitionParser("publisher", new PublisherParser());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.config.xml;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.beans.factory.config.RuntimeBeanReference;
|
||||
import org.springframework.beans.factory.support.AbstractBeanDefinition;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
|
||||
import org.springframework.beans.factory.support.ManagedMap;
|
||||
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
|
||||
import org.springframework.beans.factory.xml.ParserContext;
|
||||
import org.springframework.integration.aop.MethodNameMappingExpressionSource;
|
||||
import org.springframework.integration.channel.ChannelResolver;
|
||||
import org.springframework.integration.channel.MapBasedChannelResolver;
|
||||
import org.springframework.integration.context.IntegrationContextUtils;
|
||||
import org.springframework.integration.core.MessageChannel;
|
||||
import org.springframework.util.StringUtils;
|
||||
import org.springframework.util.xml.DomUtils;
|
||||
import org.w3c.dom.Element;
|
||||
|
||||
/**
|
||||
* @author Oleg Zhurakousky
|
||||
* @since 2.0
|
||||
*/
|
||||
public class PublisherParser extends AbstractBeanDefinitionParser {
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) {
|
||||
BeanDefinitionBuilder rootBuilder = BeanDefinitionBuilder.genericBeanDefinition(
|
||||
IntegrationNamespaceUtils.BASE_PACKAGE + ".aop.MessagePublishingInterceptor");
|
||||
|
||||
BeanDefinitionBuilder spelSourceBilder = BeanDefinitionBuilder.genericBeanDefinition(MethodNameMappingExpressionSource.class.getName());
|
||||
Map<String, Map<?,?>> mappings = this.getMappings(element, element.getAttribute("default-channel"));
|
||||
spelSourceBilder.addConstructorArgValue(mappings.get("payload"));
|
||||
MethodNameMappingExpressionSource m = null;
|
||||
if (mappings.get("headers") != null){
|
||||
spelSourceBilder.addPropertyValue("headerExpressionMap", mappings.get("headers"));
|
||||
}
|
||||
BeanDefinitionBuilder chResolverBuilder = BeanDefinitionBuilder.genericBeanDefinition(MapBasedChannelResolver.class.getName());
|
||||
if (mappings.get("channels") != null){
|
||||
spelSourceBilder.addPropertyValue("channelMap", mappings.get("channels"));
|
||||
chResolverBuilder.addConstructorArgValue(mappings.get("resolvableChannels"));
|
||||
}
|
||||
String chResolverName =
|
||||
BeanDefinitionReaderUtils.registerWithGeneratedName(chResolverBuilder.getBeanDefinition(), parserContext.getRegistry());
|
||||
String defaultChannel = StringUtils.hasText(element.getAttribute("default-channel")) ?
|
||||
element.getAttribute("default-channel") : IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME;
|
||||
|
||||
String spelSourceName =
|
||||
BeanDefinitionReaderUtils.registerWithGeneratedName(spelSourceBilder.getBeanDefinition(), parserContext.getRegistry());
|
||||
rootBuilder.addConstructorArgReference(spelSourceName);
|
||||
|
||||
rootBuilder.addPropertyReference("channelResolver", chResolverName);
|
||||
|
||||
rootBuilder.addPropertyReference("defaultChannel", defaultChannel);
|
||||
|
||||
return rootBuilder.getBeanDefinition();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private Map<String,Map<?,?>> getMappings(Element element, String defaultChannel){
|
||||
List<Element> mappings = DomUtils.getChildElementsByTagName(element, "method");
|
||||
Map<String, Map<?,?>> interceptorMappings = new HashMap<String, Map<?,?>>();
|
||||
Map<String, String> payloadExpressionMap = new HashMap<String, String>();
|
||||
Map<String, String[]> headersExpressionMap = new HashMap<String, String[]>();
|
||||
Map<String, String> channelMap = new HashMap<String, String>();
|
||||
ManagedMap resolvableChannelMap = new ManagedMap();
|
||||
if (mappings != null && mappings.size() > 0){
|
||||
for (Element mapping : mappings) {
|
||||
// set payloadMap
|
||||
String methodPattern = StringUtils.hasText(mapping.getAttribute("pattern")) ?
|
||||
mapping.getAttribute("pattern") : "*" ;
|
||||
String payloadExpression = StringUtils.hasText(mapping.getAttribute("payload")) ?
|
||||
mapping.getAttribute("payload") : "#return" ;
|
||||
payloadExpressionMap.put(methodPattern, payloadExpression);
|
||||
// set headersMap
|
||||
String headersExpression = mapping.getAttribute("headers");
|
||||
if (StringUtils.hasText(headersExpression)){
|
||||
headersExpressionMap.put(methodPattern, StringUtils.commaDelimitedListToStringArray(headersExpression));
|
||||
}
|
||||
// set channelMap
|
||||
String tmpChannel = mapping.getAttribute("channel");
|
||||
String channel = StringUtils.hasText(tmpChannel) ? tmpChannel : defaultChannel;
|
||||
channelMap.put(methodPattern, channel);
|
||||
resolvableChannelMap.put(channel, new RuntimeBeanReference(channel));
|
||||
}
|
||||
}
|
||||
|
||||
if (payloadExpressionMap.size() == 0){
|
||||
payloadExpressionMap.put("*", "#return");
|
||||
}
|
||||
interceptorMappings.put("payload", payloadExpressionMap);
|
||||
|
||||
if (headersExpressionMap.size() > 0){
|
||||
interceptorMappings.put("headers", headersExpressionMap);
|
||||
}
|
||||
if (channelMap.size() > 0){
|
||||
interceptorMappings.put("channels", channelMap);
|
||||
interceptorMappings.put("resolvableChannels", resolvableChannelMap);
|
||||
}
|
||||
return interceptorMappings;
|
||||
}
|
||||
}
|
||||
@@ -1476,5 +1476,34 @@
|
||||
</xsd:extension>
|
||||
</xsd:complexContent>
|
||||
</xsd:complexType>
|
||||
|
||||
<xsd:element name="publisher">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Defines a MessagePublishingInterceptor which allows you to generate messages as a by-product of
|
||||
method invocations on Spring configured components.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:complexType>
|
||||
<xsd:sequence>
|
||||
<xsd:element name="method" minOccurs="0" maxOccurs="unbounded">
|
||||
<xsd:complexType>
|
||||
<xsd:attribute name="pattern" type="xsd:string" />
|
||||
<xsd:attribute name="payload" type="xsd:string" />
|
||||
<xsd:attribute name="headers" type="xsd:string" />
|
||||
<xsd:attribute name="channel" type="xsd:string" />
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
</xsd:sequence>
|
||||
<xsd:attribute name="id" type="xsd:ID" use="required" />
|
||||
<xsd:attribute name="default-channel" type="xsd:string" default="nullChannel">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
Specifies default-channel to publish messages
|
||||
]]></xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
</xsd:schema>
|
||||
@@ -0,0 +1,34 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans:beans xmlns="http://www.springframework.org/schema/integration"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns:aop="http://www.springframework.org/schema/aop"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd
|
||||
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
|
||||
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd"
|
||||
xmlns:beans="http://www.springframework.org/schema/beans">
|
||||
|
||||
<beans:bean id="testBean"
|
||||
class="org.springframework.integration.config.xml.MessagePublishingInterceptorParserTests$TestBean" />
|
||||
<beans:bean id="defaultTestBean"
|
||||
class="org.springframework.integration.config.xml.MessagePublishingInterceptorParserTests$DefaultTestBean" />
|
||||
|
||||
<publish-subscribe-channel id="defaultChannel"/>
|
||||
<publish-subscribe-channel id="echoChannel"/>
|
||||
<publish-subscribe-channel id="echoUpperCaseChannel"/>
|
||||
|
||||
<aop:config>
|
||||
<aop:advisor advice-ref="interceptor" pointcut="bean(testBean)" />
|
||||
</aop:config>
|
||||
|
||||
<publisher id="interceptor" default-channel="defaultChannel">
|
||||
<method pattern="echo" payload="'Echoing: ' + #return" headers="foo='bar'" channel="echoChannel"/>
|
||||
<method pattern="echoDef*" payload="#return"/>
|
||||
</publisher>
|
||||
|
||||
<aop:config>
|
||||
<aop:advisor advice-ref="anotherInterceptor" pointcut="bean(defaultTestBean)" />
|
||||
</aop:config>
|
||||
|
||||
<publisher id="anotherInterceptor"/>
|
||||
|
||||
</beans:beans>
|
||||
@@ -0,0 +1,113 @@
|
||||
/*
|
||||
* Copyright 2002-2008 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.config.xml;
|
||||
|
||||
import static junit.framework.Assert.assertEquals;
|
||||
import static org.mockito.Matchers.anyObject;
|
||||
import static org.mockito.Mockito.doAnswer;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.mockito.Mockito;
|
||||
import org.mockito.invocation.InvocationOnMock;
|
||||
import org.mockito.stubbing.Answer;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.integration.channel.SubscribableChannel;
|
||||
import org.springframework.integration.core.Message;
|
||||
import org.springframework.integration.message.MessageHandler;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
|
||||
/**
|
||||
* @author Oleg Zhurakousky
|
||||
* @since 2.0
|
||||
*/
|
||||
@ContextConfiguration
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
public class MessagePublishingInterceptorParserTests {
|
||||
|
||||
@Autowired
|
||||
private TestBean testBean;
|
||||
@Autowired
|
||||
private DefaultTestBean defaultTestBean;
|
||||
@Autowired
|
||||
@Qualifier("defaultChannel")
|
||||
private SubscribableChannel defaultChannel;
|
||||
@Autowired
|
||||
@Qualifier("echoChannel")
|
||||
private SubscribableChannel echoChannel;
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Test
|
||||
public void validateDefaultChannelPublishing(){
|
||||
MessageHandler handler = Mockito.mock(MessageHandler.class);
|
||||
defaultChannel.subscribe(handler);
|
||||
doAnswer(new Answer() {
|
||||
public Object answer(InvocationOnMock invocation) {
|
||||
Message<?> message = (Message<?>) invocation.getArguments()[0];
|
||||
assertEquals("hello",message.getPayload());
|
||||
return null;
|
||||
}})
|
||||
.when(handler).handleMessage((Message<?>) anyObject());
|
||||
|
||||
testBean.echoDefaultChannel("hello");
|
||||
verify(handler, times(1)).handleMessage((Message<?>) anyObject());
|
||||
}
|
||||
@SuppressWarnings("unchecked")
|
||||
@Test
|
||||
public void validateEchoChannelPublishing(){
|
||||
MessageHandler handler = Mockito.mock(MessageHandler.class);
|
||||
echoChannel.subscribe(handler);
|
||||
doAnswer(new Answer() {
|
||||
public Object answer(InvocationOnMock invocation) {
|
||||
Message<?> message = (Message<?>) invocation.getArguments()[0];
|
||||
assertEquals("bar", message.getHeaders().get("foo"));
|
||||
assertEquals("Echoing: hello", message.getPayload());
|
||||
return null;
|
||||
}})
|
||||
.when(handler).handleMessage((Message<?>) anyObject());
|
||||
|
||||
testBean.echo("hello");
|
||||
verify(handler, times(1)).handleMessage((Message<?>) anyObject());
|
||||
}
|
||||
/**
|
||||
* Need to set 'debug' level
|
||||
*/
|
||||
@Test
|
||||
public void validateNullChannelPublishing(){
|
||||
defaultTestBean.echo("hello");
|
||||
}
|
||||
|
||||
public static class TestBean{
|
||||
public String echo(String str){
|
||||
return str;
|
||||
}
|
||||
public String echoUpperCase(String str){
|
||||
return str.toUpperCase();
|
||||
}
|
||||
public String echoDefaultChannel(String str){
|
||||
return str;
|
||||
}
|
||||
}
|
||||
public static class DefaultTestBean{
|
||||
public String echo(String str){
|
||||
return str;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user