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

Added test with MessageSelectors for the MessageReceivingExecutors.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-10 03:25:07 +00:00
parent a3ceb15cf1
commit 418d3f49b8
1 changed files with 26 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java
View File

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