diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractTargetEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractTargetEndpointParser.java index f9e21c83db..1b9d8d83a0 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractTargetEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractTargetEndpointParser.java @@ -24,7 +24,6 @@ import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; @@ -46,16 +45,14 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef private static final String SUBSCRIPTION_PROPERTY = "subscription"; - private static final String SELECTOR_ELEMENT = "selector"; - - private static final String REF_ATTRIBUTE = "ref"; - - private static final String SELECTORS_PROPERTY = "messageSelectors"; - private static final String ERROR_HANDLER_ATTRIBUTE = "error-handler"; private static final String ERROR_HANDLER_PROPERTY = "errorHandler"; + private static final String SELECTOR_ATTRIBUTE = "selector"; + + private static final String SELECTOR_PROPERTY = "messageSelector"; + private static final String PERIOD_ATTRIBUTE = "period"; private static final String SCHEDULE_ELEMENT = "schedule"; @@ -90,7 +87,6 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef this.parseTarget(element, this.getTargetAttributeName(), parserContext, builder); String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE); Schedule schedule = null; - ManagedList selectors = new ManagedList(); NodeList childNodes = element.getChildNodes(); for (int i = 0; i < childNodes.getLength(); i++) { Node child = childNodes.item(i); @@ -99,10 +95,6 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef if (CONCURRENCY_ELEMENT.equals(localName)) { parseConcurrencyPolicy((Element) child, builder); } - else if (SELECTOR_ELEMENT.equals(localName)) { - String ref = ((Element) child).getAttribute(REF_ATTRIBUTE); - selectors.add(new RuntimeBeanReference(ref)); - } else if (SCHEDULE_ELEMENT.equals(localName)) { schedule = this.parseSchedule((Element) child); } @@ -118,13 +110,14 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName)); builder.addPropertyReference(SUBSCRIPTION_PROPERTY, subscriptionBeanName); } - if (selectors.size() > 0) { - builder.addPropertyValue(SELECTORS_PROPERTY, selectors); - } String errorHandlerRef = element.getAttribute(ERROR_HANDLER_ATTRIBUTE); if (StringUtils.hasText(errorHandlerRef)) { builder.addPropertyReference(ERROR_HANDLER_PROPERTY, errorHandlerRef); } + String selectorRef = element.getAttribute(SELECTOR_ATTRIBUTE); + if (StringUtils.hasText(selectorRef)) { + builder.addPropertyReference(SELECTOR_PROPERTY, selectorRef); + } this.postProcess(builder, element); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index ef7ddd0e97..b4bb1a4b9c 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -235,17 +235,6 @@ - - - - - Provides a message selector reference. - - - - - - @@ -378,10 +367,10 @@ - + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java index b69753cd27..3d97f44276 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java @@ -16,10 +16,8 @@ package org.springframework.integration.endpoint; -import java.util.List; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.ThreadPoolExecutor; @@ -55,7 +53,7 @@ public class TargetEndpoint extends AbstractEndpoint implements Target, ChannelR private volatile ErrorHandler errorHandler; - private final List selectors = new CopyOnWriteArrayList(); + private volatile MessageSelector selector; private volatile ChannelRegistry channelRegistry; @@ -82,14 +80,8 @@ public class TargetEndpoint extends AbstractEndpoint implements Target, ChannelR this.target = target; } - public void setMessageSelectors(List selectors) { - this.selectors.clear(); - this.selectors.addAll(selectors); - } - - public void addMessageSelector(MessageSelector messageSelector) { - Assert.notNull(messageSelector, "'messageSelector' must not be null"); - this.selectors.add(messageSelector); + public void setMessageSelector(MessageSelector selector) { + this.selector = selector; } public Subscription getSubscription() { @@ -184,10 +176,8 @@ public class TargetEndpoint extends AbstractEndpoint implements Target, ChannelR if (!this.isRunning()) { throw new MessageHandlerNotRunningException(message); } - for (MessageSelector selector : this.selectors) { - if (!selector.accept(message)) { - return false; - } + if (this.selector != null && !this.selector.accept(message)) { + return false; } try { return this.target.send(message); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/selector/MessageSelectorChain.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/selector/MessageSelectorChain.java new file mode 100644 index 0000000000..df2957b38a --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/selector/MessageSelectorChain.java @@ -0,0 +1,72 @@ +/* + * 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.message.selector; + +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.springframework.integration.message.Message; + +/** + * A message selector implementation that passes incoming messages through a + * chain of selectors. The chain will be broken by any selector that returns + * false. + * + * @author Mark Fisher + */ +public class MessageSelectorChain implements MessageSelector { + + private final List selectors = new CopyOnWriteArrayList(); + + + /** + * Add a selector to the end of the chain. + */ + public void add(MessageSelector selector) { + this.selectors.add(selector); + } + + /** + * Add a selector to the chain at the specified index. + */ + public void add(int index, MessageSelector selector) { + this.selectors.add(index, selector); + } + + /** + * Initialize the selector chain. Removes any existing selectors. + */ + public void setSelectors(List selectors) { + this.selectors.clear(); + this.selectors.addAll(selectors); + } + + /** + * Pass the message through the selector chain. As soon as a + * selector returns 'false', this method will return 'false'. + * If all selectors accept, this method will return 'true'. + */ + public final boolean accept(Message message) { + for (MessageSelector next : this.selectors) { + if (!next.accept(message)) { + return false; + } + } + return true; + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java index 36b3596c0d..52c3ab4692 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java @@ -369,8 +369,8 @@ public class SubscriptionManagerTests { SubscriptionManager manager = new SubscriptionManager(channel, scheduler); HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2); - endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class)); - endpoint2.addMessageSelector(new PayloadTypeSelector(String.class)); + endpoint1.setMessageSelector(new PayloadTypeSelector(Integer.class)); + endpoint2.setMessageSelector(new PayloadTypeSelector(String.class)); manager.addTarget(endpoint1); manager.addTarget(endpoint2); manager.start(); @@ -395,7 +395,7 @@ public class SubscriptionManagerTests { SubscriptionManager manager = new SubscriptionManager(channel, scheduler); final HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); final HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2); - endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class) { + endpoint1.setMessageSelector(new PayloadTypeSelector(Integer.class) { @Override public boolean accept(Message message) { selectorCounter1.incrementAndGet(); @@ -403,7 +403,7 @@ public class SubscriptionManagerTests { return super.accept(message); } }); - endpoint2.addMessageSelector(new PayloadTypeSelector(Integer.class) { + endpoint2.setMessageSelector(new PayloadTypeSelector(Integer.class) { @Override public boolean accept(Message message) { selectorCounter2.incrementAndGet(); @@ -437,8 +437,8 @@ public class SubscriptionManagerTests { endpoint1.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2); endpoint2.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1)); - endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class)); - endpoint2.addMessageSelector(new PayloadTypeSelector(String.class)); + endpoint1.setMessageSelector(new PayloadTypeSelector(Integer.class)); + endpoint2.setMessageSelector(new PayloadTypeSelector(String.class)); manager.addTarget(endpoint1); manager.addTarget(endpoint2); manager.start(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java index 6496db0003..8dc8b7bbd6 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import java.util.concurrent.TimeUnit; @@ -107,13 +108,13 @@ public class EndpointParserTests { @Test public void testEndpointWithSelectorAccepts() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "endpointWithSelectors.xml", this.getClass()); + "endpointWithSelector.xml", this.getClass()); Target endpoint = (Target) context.getBean("endpoint"); ((Lifecycle) endpoint).start(); Message message = new StringMessage("test"); MessageChannel replyChannel = new QueueChannel(); message.getHeader().setReturnAddress(replyChannel); - endpoint.send(message); + assertTrue(endpoint.send(message)); Message reply = replyChannel.receive(500); assertNotNull(reply); assertEquals("foo", reply.getPayload()); @@ -122,10 +123,13 @@ public class EndpointParserTests { @Test public void testEndpointWithSelectorRejects() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "endpointWithSelectors.xml", this.getClass()); + "endpointWithSelector.xml", this.getClass()); Target endpoint = (Target) context.getBean("endpoint"); ((Lifecycle) endpoint).start(); - assertFalse(endpoint.send(new GenericMessage(123))); + Message message = new GenericMessage(123); + MessageChannel replyChannel = new QueueChannel(); + message.getHeader().setReturnAddress(replyChannel); + assertFalse(endpoint.send(message)); } @Test diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestHandler.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestHandler.java index 9a38a8f355..2855155dd1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestHandler.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestHandler.java @@ -48,7 +48,7 @@ public class TestHandler implements MessageHandler { } public Message handle(Message message) { - this.messageString = (String) message.getPayload(); + this.messageString = message.getPayload().toString(); this.latch.countDown(); return (this.replyMessageText != null) ? new StringMessage(this.replyMessageText) : null; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithSelectors.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithSelector.xml similarity index 88% rename from org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithSelectors.xml rename to org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithSelector.xml index 1a94e5658b..be3af7efb4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithSelectors.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithSelector.xml @@ -11,9 +11,9 @@ - + - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java index c53063031a..48e2a08c00 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java @@ -39,6 +39,7 @@ 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.message.selector.MessageSelectorChain; import org.springframework.integration.util.ErrorHandler; /** @@ -338,7 +339,7 @@ public class HandlerEndpointTests { @Test public void testEndpointWithSelectorRejecting() { HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.nullHandler()); - endpoint.addMessageSelector(new MessageSelector() { + endpoint.setMessageSelector(new MessageSelector() { public boolean accept(Message message) { return false; } @@ -351,7 +352,7 @@ public class HandlerEndpointTests { public void testEndpointWithSelectorAccepting() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countDownHandler(latch)); - endpoint.addMessageSelector(new MessageSelector() { + endpoint.setMessageSelector(new MessageSelector() { public boolean accept(Message message) { return true; } @@ -367,18 +368,20 @@ public class HandlerEndpointTests { public void testEndpointWithMultipleSelectorsAndFirstRejects() { final AtomicInteger counter = new AtomicInteger(); HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(counter)); - endpoint.addMessageSelector(new MessageSelector() { + MessageSelectorChain selectorChain = new MessageSelectorChain(); + selectorChain.add(new MessageSelector() { public boolean accept(Message message) { counter.incrementAndGet(); return false; } }); - endpoint.addMessageSelector(new MessageSelector() { + selectorChain.add(new MessageSelector() { public boolean accept(Message message) { counter.incrementAndGet(); return true; } }); + endpoint.setMessageSelector(selectorChain); endpoint.start(); assertFalse(endpoint.send(new StringMessage("test"))); assertEquals("only the first selector should have been invoked", 1, counter.get()); @@ -390,18 +393,20 @@ public class HandlerEndpointTests { final AtomicInteger selectorCounter = new AtomicInteger(); AtomicInteger handlerCounter = new AtomicInteger(); HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(handlerCounter)); - endpoint.addMessageSelector(new MessageSelector() { + MessageSelectorChain selectorChain = new MessageSelectorChain(); + selectorChain.add(new MessageSelector() { public boolean accept(Message message) { selectorCounter.incrementAndGet(); return true; } }); - endpoint.addMessageSelector(new MessageSelector() { + selectorChain.add(new MessageSelector() { public boolean accept(Message message) { selectorCounter.incrementAndGet(); return false; } }); + endpoint.setMessageSelector(selectorChain); endpoint.start(); assertFalse(endpoint.send(new StringMessage("test"))); assertEquals("both selectors should have been invoked", 2, selectorCounter.get()); @@ -413,18 +418,20 @@ public class HandlerEndpointTests { public void testEndpointWithMultipleSelectorsAndBothAccept() { final AtomicInteger counter = new AtomicInteger(); HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(counter)); - endpoint.addMessageSelector(new MessageSelector() { + MessageSelectorChain selectorChain = new MessageSelectorChain(); + selectorChain.add(new MessageSelector() { public boolean accept(Message message) { counter.incrementAndGet(); return true; } }); - endpoint.addMessageSelector(new MessageSelector() { + selectorChain.add(new MessageSelector() { public boolean accept(Message message) { counter.incrementAndGet(); return true; } }); + endpoint.setMessageSelector(selectorChain); endpoint.start(); assertTrue(endpoint.send(new StringMessage("test"))); assertEquals("both selectors and handler should have been invoked", 3, counter.get());