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

Moved MessageSelector interface and implementations into a new 'org.springframework.integration.selector' package (instead of a sub-package under 'message'). Also added support for "Strategy" enum in MessageSelector with the following values available [ALL, ANY, AT_LEAST_HALF, MORE_THAN_HALF] (INT-308).

Browse Source
This commit is contained in:
Mark Fisher
2008-10-14 16:16:55 +00:00
parent 9286f1fdaa
commit 109343d14e
37 changed files with 372 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelPurger.java
View File

@@ -20,7 +20,7 @@ import java.util.ArrayList;
import java.util.List;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.util.Assert;
/**

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/MessageChannelTemplate.java
View File

@@ -26,7 +26,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/NullChannel.java
View File

@@ -22,7 +22,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
/**
* A channel implementation that essentially behaves like "/dev/null".

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannel.java
View File

@@ -19,7 +19,7 @@ package org.springframework.integration.channel;
import java.util.List;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
/**
* @author Mark Fisher

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/QueueChannel.java
View File

@@ -23,7 +23,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.util.Assert;
/**

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/ThreadLocalChannel.java
View File

@@ -22,7 +22,7 @@ import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
/**
* A channel implementation that stores messages in a thread-bound queue. In

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptor.java
View File

@@ -22,7 +22,7 @@ import java.util.List;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
/**
* A {@link org.springframework.integration.channel.ChannelInterceptor} that

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java
View File

@@ -24,7 +24,7 @@ import org.springframework.integration.channel.BlockingChannel;
import org.springframework.integration.channel.ChannelInterceptor;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.util.Assert;
/**

2
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractChannelParser.java
View File

@@ -27,7 +27,7 @@ import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.channel.interceptor.MessageSelectingInterceptor;
import org.springframework.integration.message.selector.PayloadTypeSelector;
import org.springframework.integration.selector.PayloadTypeSelector;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;

2
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/SelectorChainParser.java
View File

@@ -24,7 +24,7 @@ import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.integration.message.selector.MessageSelectorChain;
import org.springframework.integration.selector.MessageSelectorChain;
/**
* Parser for the <selector-chain/> element.

2
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java
View File

@@ -29,7 +29,7 @@ import org.springframework.integration.message.MessageHeaders;
import org.springframework.integration.message.MessageProducer;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.util.Assert;
/**

2
org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java
View File

@@ -19,7 +19,7 @@ package org.springframework.integration.filter;
import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
import org.springframework.integration.endpoint.ReplyMessageHolder;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.util.Assert;
/**

2
org.springframework.integration/src/main/java/org/springframework/integration/filter/MethodInvokingSelector.java
View File

@@ -20,7 +20,7 @@ import java.lang.reflect.Method;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageMappingMethodInvoker;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.util.Assert;
/**

2
org.springframework.integration/src/main/java/org/springframework/integration/message/selector/MessageSelector.java → org.springframework.integration/src/main/java/org/springframework/integration/selector/MessageSelector.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.message.selector;
package org.springframework.integration.selector;
import org.springframework.integration.message.Message;

51
org.springframework.integration/src/main/java/org/springframework/integration/message/selector/MessageSelectorChain.java → org.springframework.integration/src/main/java/org/springframework/integration/selector/MessageSelectorChain.java
View File

@@ -14,12 +14,13 @@
* limitations under the License.
*/
package org.springframework.integration.message.selector;
package org.springframework.integration.selector;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import org.springframework.integration.message.Message;
import org.springframework.util.Assert;
/**
* A message selector implementation that passes incoming messages through a
@@ -30,9 +31,19 @@ import org.springframework.integration.message.Message;
*/
public class MessageSelectorChain implements MessageSelector {
public static enum Strategy { ALL, MORE_THAN_HALF, AT_LEAST_HALF, ANY };
private volatile Strategy strategy = Strategy.ALL;
private final List<MessageSelector> selectors = new CopyOnWriteArrayList<MessageSelector>();
public void setStrategy(Strategy strategy) {
Assert.notNull(strategy, "strategy must not be null");
this.strategy = strategy;
}
/**
* Add a selector to the end of the chain.
*/
@@ -51,8 +62,11 @@ public class MessageSelectorChain implements MessageSelector {
* Initialize the selector chain. Removes any existing selectors.
*/
public void setSelectors(List<MessageSelector> selectors) {
this.selectors.clear();
this.selectors.addAll(selectors);
Assert.notEmpty(selectors, "selectors must not be empty");
synchronized (this.selectors) {
this.selectors.clear();
this.selectors.addAll(selectors);
}
}
/**
@@ -61,12 +75,39 @@ public class MessageSelectorChain implements MessageSelector {
* If all selectors accept, this method will return 'true'.
*/
public final boolean accept(Message<?> message) {
int count = 0;
int accepted = 0;
for (MessageSelector next : this.selectors) {
if (!next.accept(message)) {
count++;
if (next.accept(message)) {
if (this.strategy.equals(Strategy.ANY)) {
return true;
}
accepted++;
}
else if (this.strategy.equals(Strategy.ALL)) {
return false;
}
}
return true;
return this.decide(accepted, count);
}
private boolean decide(int accepted, int total) {
if (accepted == 0) {
return false;
}
switch (this.strategy) {
case ANY:
return true;
case ALL:
return (accepted == total);
case MORE_THAN_HALF:
return (2 * accepted) > total;
case AT_LEAST_HALF:
return (2 * accepted) >= total;
default:
throw new IllegalArgumentException("unsupported strategy " + this.strategy);
}
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/message/selector/PayloadTypeSelector.java → org.springframework.integration/src/main/java/org/springframework/integration/selector/PayloadTypeSelector.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.message.selector;
package org.springframework.integration.selector;
import java.util.ArrayList;
import java.util.List;

2
org.springframework.integration/src/main/java/org/springframework/integration/message/selector/UnexpiredMessageSelector.java → org.springframework.integration/src/main/java/org/springframework/integration/selector/UnexpiredMessageSelector.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.message.selector;
package org.springframework.integration.selector;
import org.springframework.integration.message.Message;

2
org.springframework.integration/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java
View File

@@ -26,7 +26,7 @@ import org.junit.Test;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
/**
* @author Mark Fisher

2
org.springframework.integration/src/test/java/org/springframework/integration/channel/QueueChannelTests.java
View File

@@ -34,7 +34,7 @@ import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.selector.UnexpiredMessageSelector;
import org.springframework.integration.selector.UnexpiredMessageSelector;
/**
* @author Mark Fisher

2
org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptorTests.java
View File

@@ -27,7 +27,7 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
/**
* @author Mark Fisher

2
org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java
View File

@@ -26,7 +26,7 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
/**
* @author Mark Fisher

2
org.springframework.integration/src/test/java/org/springframework/integration/config/FilterParserTests.java
View File

@@ -29,7 +29,7 @@ import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.StringUtils;

4
org.springframework.integration/src/test/java/org/springframework/integration/config/SelectorChainParserTests.java
View File

@@ -24,8 +24,8 @@ import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.message.selector.MessageSelectorChain;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelectorChain;
/**
* @author Mark Fisher

2
org.springframework.integration/src/test/java/org/springframework/integration/config/StubMessageSelector.java
View File

@@ -17,7 +17,7 @@
package org.springframework.integration.config;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
/**
* @author Mark Fisher

2
org.springframework.integration/src/test/java/org/springframework/integration/config/TestSelector.java
View File

@@ -17,7 +17,7 @@
package org.springframework.integration.config;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
/**
* @author Mark Fisher

2
org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithSelector.xml
View File

@@ -15,7 +15,7 @@
ref="testHandler" selector="typeSelector">
</service-activator>
<beans:bean id="typeSelector" class="org.springframework.integration.message.selector.PayloadTypeSelector">
<beans:bean id="typeSelector" class="org.springframework.integration.selector.PayloadTypeSelector">
<beans:constructor-arg value="java.lang.String"/>
</beans:bean>

2
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java
View File

@@ -34,7 +34,7 @@ import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.TestHandlers;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
/**
* @author Mark Fisher

4
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ServiceActivatorEndpointTests.java
View File

@@ -37,8 +37,8 @@ import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.TestHandlers;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.message.selector.MessageSelectorChain;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelectorChain;
/**
* @author Mark Fisher

2
org.springframework.integration/src/test/java/org/springframework/integration/filter/MessageFilterTests.java
View File

@@ -29,7 +29,7 @@ import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.filter.MessageFilter;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.selector.MessageSelector;
/**
* @author Mark Fisher

1
org.springframework.integration/src/test/java/org/springframework/integration/message/selector/PayloadTypeSelectorTests.java
View File

@@ -25,6 +25,7 @@ import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.selector.PayloadTypeSelector;
/**
* @author Mark Fisher

1
org.springframework.integration/src/test/java/org/springframework/integration/message/selector/UnexpiredMessageSelectorTests.java
View File

@@ -23,6 +23,7 @@ import org.junit.Test;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.selector.UnexpiredMessageSelector;
/**
* @author Mark Fisher

175
org.springframework.integration/src/test/java/org/springframework/integration/selector/MessageSelectorChainTests.java Normal file
View File

@@ -0,0 +1,175 @@
/*
* 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.selector;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
/**
* @author Mark Fisher
*/
public class MessageSelectorChainTests {
private final Message<?> message = new StringMessage("test");
@Test
public void anyStrategyAccepts() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.ANY);
chain.add(new TestSelector(false));
chain.add(new TestSelector(false));
chain.add(new TestSelector(true));
chain.add(new TestSelector(false));
assertTrue(chain.accept(message));
}
@Test
public void anyStrategyRejects() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.ANY);
chain.add(new TestSelector(false));
chain.add(new TestSelector(false));
chain.add(new TestSelector(false));
chain.add(new TestSelector(false));
assertFalse(chain.accept(message));
}
@Test
public void allStrategyAccepts() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.ALL);
chain.add(new TestSelector(true));
chain.add(new TestSelector(true));
chain.add(new TestSelector(true));
assertTrue(chain.accept(message));
}
@Test
public void allStrategyRejects() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.ALL);
chain.add(new TestSelector(true));
chain.add(new TestSelector(false));
chain.add(new TestSelector(true));
assertFalse(chain.accept(message));
}
@Test
public void atLeastHalfStrategyWithOddNumberAccepts() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.AT_LEAST_HALF);
chain.add(new TestSelector(true));
chain.add(new TestSelector(true));
chain.add(new TestSelector(false));
assertTrue(chain.accept(message));
}
@Test
public void atLeastHalfStrategyWithEvenNumberAccepts() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.AT_LEAST_HALF);
chain.add(new TestSelector(true));
chain.add(new TestSelector(true));
chain.add(new TestSelector(false));
chain.add(new TestSelector(false));
assertTrue(chain.accept(message));
}
@Test
public void atLeastHalfStrategyWithOddNumberRejects() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.AT_LEAST_HALF);
chain.add(new TestSelector(false));
chain.add(new TestSelector(true));
chain.add(new TestSelector(false));
assertFalse(chain.accept(message));
}
@Test
public void atLeastHalfStrategyWithEvenNumberRejects() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.AT_LEAST_HALF);
chain.add(new TestSelector(false));
chain.add(new TestSelector(true));
chain.add(new TestSelector(false));
chain.add(new TestSelector(false));
assertFalse(chain.accept(message));
}
@Test
public void moreThanHalfStrategyWithOddNumberAccepts() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.MORE_THAN_HALF);
chain.add(new TestSelector(true));
chain.add(new TestSelector(true));
chain.add(new TestSelector(false));
assertTrue(chain.accept(message));
}
@Test
public void moreThanHalfStrategyWithEvenNumberAccepts() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.MORE_THAN_HALF);
chain.add(new TestSelector(true));
chain.add(new TestSelector(true));
chain.add(new TestSelector(false));
chain.add(new TestSelector(true));
assertTrue(chain.accept(message));
}
@Test
public void moreThanHalfStrategyWithOddNumberRejects() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.MORE_THAN_HALF);
chain.add(new TestSelector(false));
chain.add(new TestSelector(true));
chain.add(new TestSelector(false));
assertFalse(chain.accept(message));
}
@Test
public void moreThanHalfStrategyWithEvenNumberRejects() {
MessageSelectorChain chain = new MessageSelectorChain();
chain.setStrategy(MessageSelectorChain.Strategy.MORE_THAN_HALF);
chain.add(new TestSelector(false));
chain.add(new TestSelector(true));
chain.add(new TestSelector(true));
chain.add(new TestSelector(false));
assertFalse(chain.accept(message));
}
private static class TestSelector implements MessageSelector {
private final boolean accept;
private TestSelector(boolean accept) {
this.accept = accept;
}
public boolean accept(Message<?> message) {
return this.accept;
}
}
}