diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java index a34c38cb42..0b96868621 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java @@ -37,12 +37,7 @@ public class ChannelParser implements BeanDefinitionParser { private static final String CAPACITY_ATTRIBUTE = "capacity"; - private boolean isBroadcaster; - - - public ChannelParser(boolean isBroadcaster) { - this.isBroadcaster = isBroadcaster; - } + private static final String PUBLISH_SUBSCRIBE_ATTRIBUTE = "publish-subscribe"; public BeanDefinition parse(Element element, ParserContext parserContext) { @@ -52,7 +47,8 @@ public class ChannelParser implements BeanDefinitionParser { if (StringUtils.hasText(capacity)) { channelDef.getConstructorArgumentValues().addGenericArgumentValue(Integer.parseInt(capacity)); } - if (this.isBroadcaster) { + String isPublishSubscribe = element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE); + if ("true".equals(isPublishSubscribe)) { channelDef.getPropertyValues().addPropertyValue("broadcaster", Boolean.TRUE); } String beanName = element.getAttribute(ID_ATTRIBUTE); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java index 9696ef737e..c3fe2fce70 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java @@ -32,7 +32,7 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport { public void init() { registerBeanDefinitionParser("message-bus", new MessageBusParser()); registerBeanDefinitionParser("annotation-driven", new AnnotationDrivenParser()); - registerBeanDefinitionParser("channel", new ChannelParser(false)); + registerBeanDefinitionParser("channel", new ChannelParser()); registerBeanDefinitionParser("source-adapter", new ChannelAdapterParser(true)); registerBeanDefinitionParser("target-adapter", new ChannelAdapterParser(false)); registerBeanDefinitionParser("endpoint", new EndpointParser()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd index 97900b37e6..fc3aabc891 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd @@ -49,6 +49,7 @@ + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java index 93235acba8..c0b1f00e31 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java @@ -16,13 +16,23 @@ package org.springframework.integration.config; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + import org.junit.Test; + import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.dispatcher.DefaultMessageDispatcher; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher @@ -30,10 +40,10 @@ import org.springframework.integration.message.GenericMessage; public class ChannelParserTests { @Test - public void testSimpleChannelWithDefaults() { + public void testChannelWithCapacity() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "channelParserTests.xml", this.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("testChannel"); + MessageChannel channel = (MessageChannel) context.getBean("capacityChannel"); for (int i = 0; i < 10; i++) { boolean result = channel.send(new GenericMessage(1, "test"), 10); assertTrue(result); @@ -41,4 +51,80 @@ public class ChannelParserTests { assertFalse(channel.send(new GenericMessage(1, "test"), 3)); } + @Test + public void testPointToPointChannelByDefault() throws InterruptedException { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "channelParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelByDefault"); + channel.send(new StringMessage("test")); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + AtomicInteger counter = new AtomicInteger(); + CountDownLatch latch = new CountDownLatch(1); + TestHandler handler1 = new TestHandler(counter, latch); + TestHandler handler2 = new TestHandler(counter, latch); + dispatcher.addHandler(handler1); + dispatcher.addHandler(handler2); + dispatcher.start(); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + assertEquals(1, counter.get()); + } + + @Test + public void testPointToPointChannelExplicitlyConfigured() throws InterruptedException { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "channelParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("pointToPointChannelExplicitlyConfigured"); + channel.send(new StringMessage("test")); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + AtomicInteger counter = new AtomicInteger(); + CountDownLatch latch = new CountDownLatch(1); + TestHandler handler1 = new TestHandler(counter, latch); + TestHandler handler2 = new TestHandler(counter, latch); + dispatcher.addHandler(handler1); + dispatcher.addHandler(handler2); + dispatcher.start(); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + assertEquals(1, counter.get()); + } + + @Test + public void testPublishSubscribeChannel() throws InterruptedException { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "channelParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("publishSubscribeChannel"); + channel.send(new StringMessage("test")); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + AtomicInteger counter = new AtomicInteger(); + CountDownLatch latch = new CountDownLatch(2); + TestHandler handler1 = new TestHandler(counter, latch); + TestHandler handler2 = new TestHandler(counter, latch); + dispatcher.addHandler(handler1); + dispatcher.addHandler(handler2); + dispatcher.start(); + latch.await(500, TimeUnit.MILLISECONDS); + assertEquals(0, latch.getCount()); + assertEquals(2, counter.get()); + } + + + private static class TestHandler implements MessageHandler { + + private AtomicInteger counter; + + private CountDownLatch latch; + + TestHandler(AtomicInteger counter, CountDownLatch latch) { + this.counter = counter; + this.latch = latch; + } + + public Message handle(Message message) { + this.counter.incrementAndGet(); + this.latch.countDown(); + return null; + } + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml index 3e14d73a70..fa89cefff3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml @@ -7,6 +7,12 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + + + + + + +