From 418d3f49b85276bf136df2bc0531fa84967dab37 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 10 Jan 2008 03:25:07 +0000 Subject: [PATCH] Added test with MessageSelectors for the MessageReceivingExecutors. --- .../bus/DefaultMessageDispatcherTests.java | 27 ++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java index 1d37483374..59b448cfd2 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java @@ -30,6 +30,7 @@ import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; +import org.springframework.integration.message.selector.PayloadTypeSelector; /** * @author Mark Fisher @@ -361,11 +362,35 @@ public class DefaultMessageDispatcherTests { dispatcher.dispatch(); latch.await(100, TimeUnit.MILLISECONDS); assertEquals("endpoint1 should have received one message", 1, counter1.get()); - assertEquals("endpoint2 should have received one message", 1, counter1.get()); + assertEquals("endpoint2 should have received one message", 1, counter2.get()); assertEquals("endpoint1 should have rejected two times", 2, rejectedCounter1.get()); assertEquals("endpoint2 should have rejected four times", 4, rejectedCounter2.get()); } + @Test + public void testTwoExecutorsWithPayloadTypeMessageSelectors() throws InterruptedException { + final AtomicInteger counter1 = new AtomicInteger(); + final AtomicInteger counter2 = new AtomicInteger(); + final CountDownLatch latch = new CountDownLatch(1); + TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); + TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); + ConsumerPolicy policy = new ConsumerPolicy(); + SimpleChannel channel = new SimpleChannel(); + channel.send(new StringMessage(1, "test")); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); + MessageReceivingExecutor executor1 = new MessageReceivingExecutor(endpoint1, 1, 1); + MessageReceivingExecutor executor2 = new MessageReceivingExecutor(endpoint2, 1, 1); + executor1.addMessageSelector(new PayloadTypeSelector(Integer.class)); + executor2.addMessageSelector(new PayloadTypeSelector(String.class)); + dispatcher.addExecutor(executor1); + dispatcher.addExecutor(executor2); + dispatcher.dispatch(); + latch.await(100, TimeUnit.MILLISECONDS); + assertEquals("endpoint1 should not have accepted the message", 0, counter1.get()); + assertEquals("endpoint2 should have accepted the message", 1, counter2.get()); + } + private static class TestEndpoint extends GenericMessageEndpoint {