DelegatingMessageSelector implements MessageSelector { @Override public boolean accept(Message message) { return matchingHeaders.entrySet() .stream() .allMatch(entry -> factories.stream() .filter(factory -> factory.supports(entry.getKey())) .findAny() .orElse(new HeaderMatchingMessageSelector.Factory()) .create(entry.getKey(), entry.getValue(), context) .accept(message)); } DelegatingMessageSelector(String selector, TestContext context); @Override boolean accept(Message message); void addMessageSelectorFactory(MessageSelectorFactory factory);  }
@Test public void testHeaderMatchingSelector() { DelegatingMessageSelector messageSelector = new DelegatingMessageSelector("operation = 'foo'", context); Message acceptMessage = new DefaultMessage("FooTest") .setHeader("operation", "foo"); Message declineMessage = new DefaultMessage("FooTest") .setHeader("operation", "foobar"); Assert.assertTrue(messageSelector.accept(acceptMessage)); Assert.assertFalse(messageSelector.accept(declineMessage)); }
@Test public void testHeaderMatchingSelectorAndOperation() { DelegatingMessageSelector messageSelector = new DelegatingMessageSelector("foo = 'bar' AND operation = 'foo'", context); Message acceptMessage = new DefaultMessage("FooTest") .setHeader("foo", "bar") .setHeader("operation", "foo"); Message declineMessage = new DefaultMessage("FooTest") .setHeader("operation", "foo"); Assert.assertTrue(messageSelector.accept(acceptMessage)); Assert.assertFalse(messageSelector.accept(declineMessage)); }
@Test public void testPayloadMatchingDelegation() { DelegatingMessageSelector messageSelector = new DelegatingMessageSelector("foo = 'bar' AND payload = 'FooTest'", context); Message acceptMessage = new DefaultMessage("FooTest") .setHeader("foo", "bar") .setHeader("operation", "foo"); Message declineMessage = new DefaultMessage("BarTest") .setHeader("foo", "bar") .setHeader("operation", "foo"); Assert.assertTrue(messageSelector.accept(acceptMessage)); Assert.assertFalse(messageSelector.accept(declineMessage)); messageSelector = new DelegatingMessageSelector("payload = 'FooTest'", context); acceptMessage = new DefaultMessage("FooTest") .setHeader("foo", "bar") .setHeader("operation", "foo"); declineMessage = new DefaultMessage("BarTest") .setHeader("operation", "foo"); Assert.assertTrue(messageSelector.accept(acceptMessage)); Assert.assertFalse(messageSelector.accept(declineMessage)); }
@Test public void testPayloadAndHeaderMatchingDelegation() { DelegatingMessageSelector messageSelector = new DelegatingMessageSelector("header:payload = 'foo' AND payload = 'foo'", context); Assert.assertTrue(messageSelector.accept(new DefaultMessage("foo") .setHeader("payload", "foo"))); Assert.assertFalse(messageSelector.accept(new DefaultMessage("foo") .setHeader("payload", "bar"))); Assert.assertFalse(messageSelector.accept(new DefaultMessage("bar") .setHeader("payload", "foo"))); }
@Test public void testCustomMessageSelectorDelegation() { Map<String, MessageSelector.MessageSelectorFactory> factories = new HashMap<>(); factories.put("customSelectorFactory", new MessageSelector.MessageSelectorFactory() { @Override public boolean supports(String key) { return key.startsWith("x:"); } @Override public MessageSelector create(String key, String value, TestContext context) { return message -> message.getHeaders().get("foo").equals(value); } }); when(resolver.resolveAll(MessageSelector.MessageSelectorFactory.class)).thenReturn(factories); context.setReferenceResolver(resolver); DelegatingMessageSelector messageSelector = new DelegatingMessageSelector("x:foo = 'bar'", context); Message acceptMessage = new DefaultMessage("FooBar") .setHeader("foo", "bar") .setHeader("operation", "foo"); Message declineMessage = new DefaultMessage("FooBar") .setHeader("foo", "bars") .setHeader("operation", "foo"); Assert.assertTrue(messageSelector.accept(acceptMessage)); Assert.assertFalse(messageSelector.accept(declineMessage)); }