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

Adds support for an <interceptor/> sub-element of <message-bus/>

Browse Source
This commit is contained in:
Marius Bogoevici
2008-06-22 17:36:45 +00:00
parent f3d2c23069
commit 490a3d9951
7 changed files with 171 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
org.springframework.integration/src/main/java/org/springframework/integration/config/MessageBusParser.java
View File

@@ -16,15 +16,12 @@
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.config.BeanDefinition;
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.ManagedList;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
@@ -33,6 +30,9 @@ import org.springframework.integration.ConfigurationException;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.bus.MessageBusAwareBeanPostProcessor;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
/**
* Parser for the <em>message-bus</em> element of the integration namespace.
@@ -51,10 +51,16 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
private static final String ERROR_CHANNEL_ATTRIBUTE = "error-channel";
private static final String DEFAULT_CONCURRENCY_ELEMENT = "default-concurrency";
private static final String DEFAULT_CONCURRENCY_PROPERTY = "defaultConcurrencyPolicy";
private static final String CHANNEL_FACTORY_ATTRIBUTE = "channel-factory";
private static final String INTERCEPTOR_ELEMENT = "interceptor";
private static final String REFERENCE_ATTRIBUTE = "ref";
private static final String INTERCEPTORS_PROPERTY = "interceptors";
@Override
@@ -96,6 +102,7 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
private void processChildElements(BeanDefinitionBuilder beanDefinition, Element element) {
NodeList childNodes = element.getChildNodes();
ManagedList interceptors = new ManagedList();
for (int i = 0; i < childNodes.getLength(); i++) {
Node child = childNodes.item(i);
if (child.getNodeType() == Node.ELEMENT_NODE) {
@@ -104,8 +111,14 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
beanDefinition.addPropertyValue(DEFAULT_CONCURRENCY_PROPERTY,
IntegrationNamespaceUtils.parseConcurrencyPolicy((Element) child));
}
if (INTERCEPTOR_ELEMENT.equals(localName)) {
interceptors.add(new RuntimeBeanReference(((Element)child).getAttribute(REFERENCE_ATTRIBUTE)));
}
}
}
if (interceptors.size() > 0) {
beanDefinition.addPropertyValue(INTERCEPTORS_PROPERTY, interceptors);
}
}
@Override

1
org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -26,6 +26,7 @@
</xsd:annotation>
<xsd:sequence>
<xsd:element name="default-concurrency" type="concurrencyType" minOccurs="0" maxOccurs="1"/>
<xsd:element ref="interceptor" minOccurs="0" maxOccurs="unbounded"/>
</xsd:sequence>
<xsd:attribute name="auto-startup" type="xsd:boolean"/>
<xsd:attribute name="auto-create-channels" type="xsd:boolean"/>

67
org.springframework.integration/src/test/java/org/springframework/integration/bus/interceptor/MessageBusInterceptorTests.java
View File

@@ -19,9 +19,8 @@ package org.springframework.integration.bus.interceptor;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.springframework.integration.bus.MessageBus;
/**
@@ -38,6 +37,12 @@ public class MessageBusInterceptorTests {
messageBus.addInterceptor(startInterceptor);
messageBus.addInterceptor(stopInterceptor);
// check the state of the interceptors
executeInterceptorsTest(messageBus, startInterceptor, stopInterceptor);
}
public static void executeInterceptorsTest(MessageBus messageBus, TestMessageBusStartInterceptor startInterceptor,
TestMessageBusStopInterceptor stopInterceptor) {
assertTrue(!messageBus.isRunning());
assertEquals(startInterceptor.getPreStartCounter().get(), 0);
assertEquals(startInterceptor.getPostStartCounter().get(), 0);
assertEquals(stopInterceptor.getPreStopCounter().get(), 0);
@@ -57,64 +62,6 @@ public class MessageBusInterceptorTests {
assertEquals(stopInterceptor.getPreStopCounter().get(), 1);
assertEquals(stopInterceptor.getPostStopCounter().get(), 1);
}
private static class TestMessageBusStartInterceptor extends MessageBusInterceptorAdapter {
private AtomicInteger preStartCounter = new AtomicInteger(0);
private AtomicInteger postStartCounter = new AtomicInteger(0);
public AtomicInteger getPreStartCounter() {
return preStartCounter;
}
public AtomicInteger getPostStartCounter() {
return postStartCounter;
}
@Override
public void preStart(MessageBus bus) {
this.preStartCounter.incrementAndGet();
assertTrue(!bus.isRunning());
}
@Override
public void postStart(MessageBus bus) {
this.postStartCounter.incrementAndGet();
assertTrue(bus.isRunning());
}
}
private static class TestMessageBusStopInterceptor extends MessageBusInterceptorAdapter {
private AtomicInteger preStopCounter = new AtomicInteger(0);
private AtomicInteger postStopCounter = new AtomicInteger(0);
public AtomicInteger getPreStopCounter() {
return preStopCounter;
}
public AtomicInteger getPostStopCounter() {
return postStopCounter;
}
@Override
public void preStop(MessageBus bus) {
this.preStopCounter.incrementAndGet();
assertTrue(bus.isRunning());
}
@Override
public void postStop(MessageBus bus) {
this.postStopCounter.incrementAndGet();
assertTrue(!bus.isRunning());
}
}
}

53
org.springframework.integration/src/test/java/org/springframework/integration/bus/interceptor/TestMessageBusStartInterceptor.java Normal file
View File

@@ -0,0 +1,53 @@
/*
* 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.bus.interceptor;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.bus.MessageBus;
/**
* @author Marius Bogoevici
*/
public class TestMessageBusStartInterceptor extends MessageBusInterceptorAdapter {
private AtomicInteger preStartCounter = new AtomicInteger(0);
private AtomicInteger postStartCounter = new AtomicInteger(0);
public AtomicInteger getPreStartCounter() {
return preStartCounter;
}
public AtomicInteger getPostStartCounter() {
return postStartCounter;
}
@Override
public void preStart(MessageBus bus) {
this.preStartCounter.incrementAndGet();
org.junit.Assert.assertTrue(!bus.isRunning());
}
@Override
public void postStart(MessageBus bus) {
this.postStartCounter.incrementAndGet();
org.junit.Assert.assertTrue(bus.isRunning());
}
}

53
org.springframework.integration/src/test/java/org/springframework/integration/bus/interceptor/TestMessageBusStopInterceptor.java Normal file
View File

@@ -0,0 +1,53 @@
/*
* 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.bus.interceptor;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.integration.bus.MessageBus;
/**
* @author Marius Bogoevici
*/
public class TestMessageBusStopInterceptor extends MessageBusInterceptorAdapter {
private AtomicInteger preStopCounter = new AtomicInteger(0);
private AtomicInteger postStopCounter = new AtomicInteger(0);
public AtomicInteger getPreStopCounter() {
return preStopCounter;
}
public AtomicInteger getPostStopCounter() {
return postStopCounter;
}
@Override
public void preStop(MessageBus bus) {
this.preStopCounter.incrementAndGet();
org.junit.Assert.assertTrue(bus.isRunning());
}
@Override
public void postStop(MessageBus bus) {
this.postStopCounter.incrementAndGet();
org.junit.Assert.assertTrue(!bus.isRunning());
}
}

25
org.springframework.integration/src/test/java/org/springframework/integration/config/MessageBusParserTests.java
View File

@@ -33,6 +33,9 @@ import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.bus.TestMessageBusAwareImpl;
import org.springframework.integration.bus.interceptor.MessageBusInterceptorTests;
import org.springframework.integration.bus.interceptor.TestMessageBusStartInterceptor;
import org.springframework.integration.bus.interceptor.TestMessageBusStopInterceptor;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.dispatcher.DirectChannel;
import org.springframework.integration.endpoint.TargetEndpoint;
@@ -64,7 +67,7 @@ public class MessageBusParserTests {
assertNotNull("bus should have created a default error channel", bus.getErrorChannel());
}
@Test(expected=ConfigurationException.class)
@Test(expected = ConfigurationException.class)
public void testAutoCreateChannelsDisabledByDefault() {
ApplicationContext context = new ClassPathXmlApplicationContext(
"messageBusWithDefaults.xml", this.getClass());
@@ -110,7 +113,7 @@ public class MessageBusParserTests {
// tries to get a reference to the message bus
assertEquals(BeanCreationException.class, e.getCause().getClass());
assertEquals(e.getBeanName(), MessageBusParser.MESSAGE_BUS_AWARE_POST_PROCESSOR_BEAN_NAME);
assertEquals(ConfigurationException.class, ((BeanCreationException) e.getCause()).getCause().getClass());
assertEquals(ConfigurationException.class, (e.getCause()).getCause().getClass());
assertEquals(((BeanCreationException) e.getCause()).getBeanName(), MessageBusParser.MESSAGE_BUS_BEAN_NAME);
}
assertTrue(exceptionThrown);
@@ -140,9 +143,9 @@ public class MessageBusParserTests {
"messageBusWithDefaultConcurrencyTests.xml", this.getClass());
TargetEndpoint endpoint2 = (TargetEndpoint) context.getBean("endpoint2");
assertEquals(14, endpoint2.getConcurrencyPolicy().getCoreSize());
assertEquals(17, endpoint2.getConcurrencyPolicy().getMaxSize());
assertEquals(17, endpoint2.getConcurrencyPolicy().getMaxSize());
}
@Test
public void testMessageBusAwareAutomaticallyAddedByNamespace() {
ApplicationContext context = new ClassPathXmlApplicationContext(
@@ -153,7 +156,7 @@ public class MessageBusParserTests {
@Test
public void testMessageBusWithChannelFactory() {
ApplicationContext context = new ClassPathXmlApplicationContext("messageBusWithChannelFactory.xml",
ApplicationContext context = new ClassPathXmlApplicationContext("messageBusWithChannelFactory.xml",
this.getClass());
assertEquals(DirectChannel.class, context.getBean("defaultTypeChannel").getClass());
assertEquals(QueueChannel.class, context.getBean("specifiedTypeChannel").getClass());
@@ -194,4 +197,16 @@ public class MessageBusParserTests {
assertEquals(SimpleMessagingTaskScheduler.class, taskExecutor.getClass());
}
@Test
public void testMessageBusWithInterceptors() {
ApplicationContext context = new ClassPathXmlApplicationContext(
"messageBusWithInterceptors.xml", this.getClass());
MessageBus messageBus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
TestMessageBusStartInterceptor startInterceptor = (TestMessageBusStartInterceptor) context.getBean(
"startInterceptor");
TestMessageBusStopInterceptor stopInterceptor = (TestMessageBusStopInterceptor) context.getBean(
"stopInterceptor");
MessageBusInterceptorTests.executeInterceptorsTest(messageBus, startInterceptor, stopInterceptor);
}
}

19
org.springframework.integration/src/test/java/org/springframework/integration/config/messageBusWithInterceptors.xml Normal file
View File

@@ -0,0 +1,19 @@
<?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:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<message-bus auto-startup="false">
<interceptor ref="startInterceptor"/>
<interceptor ref="stopInterceptor"/>
</message-bus>
<beans:bean id="startInterceptor" class="org.springframework.integration.bus.interceptor.TestMessageBusStartInterceptor"/>
<beans:bean id="stopInterceptor" class="org.springframework.integration.bus.interceptor.TestMessageBusStopInterceptor"/>
</beans:beans>