Added MessageSelectorChain and modified TargetEndpoint to provide a 'setMessageSelector' method rather than managing a list of selectors with the 'addMessageSelector' method. Also modified the endpoint parser so that the configuration of a selector chain is more consistent with that of MessageHandlerChain. Now, the XML for endpoints uses a "selector" attribute instead of 0..n <selector/> sub-elements (INT-159).

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);
 	}
 

org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd

@@ -235,17 +235,6 @@
 		</xsd:complexType>
 	</xsd:element>
 
-	<xsd:element name="selector">
-		<xsd:complexType>
-			<xsd:annotation>
-				<xsd:documentation>
-	Provides a message selector reference.
-				</xsd:documentation>
-			</xsd:annotation>
-			<xsd:attribute name="ref" type="xsd:string" use="required"/>
-		</xsd:complexType>
-	</xsd:element>
-
 	<xsd:element name="handler-chain">
 		<xsd:complexType>
 			<xsd:annotation>
@@ -378,10 +367,10 @@
 				<xsd:sequence>
 					<xsd:element ref="schedule" minOccurs="0" maxOccurs="1"/>
 					<xsd:element name="concurrency" type="concurrencyType" minOccurs="0" maxOccurs="1"/>
-					<xsd:element ref="selector" minOccurs="0" maxOccurs="unbounded"/>
 				</xsd:sequence>
 				<xsd:attribute name="input-channel" type="xsd:string" use="required"/>
 				<xsd:attribute name="error-handler" type="xsd:string"/>
+				<xsd:attribute name="selector" type="xsd:string"/>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>

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<MessageSelector> selectors = new CopyOnWriteArrayList<MessageSelector>();
+	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<MessageSelector> 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);

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
+ * <em>false</em>.
+ * 
+ * @author Mark Fisher
+ */
+public class MessageSelectorChain implements MessageSelector {
+
+	private final List<MessageSelector> selectors = new CopyOnWriteArrayList<MessageSelector>();
+
+
+	/**
+	 * 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<MessageSelector> 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;
+	}
+
+}

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();

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<Integer>(123)));
+		Message<?> message = new GenericMessage<Integer>(123);
+		MessageChannel replyChannel = new QueueChannel();
+		message.getHeader().setReturnAddress(replyChannel);
+		assertFalse(endpoint.send(message));
 	}
 
 	@Test

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;
 	}

org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithSelector.xml

@@ -11,9 +11,9 @@
 
 	<queue-channel id="testChannel" capacity="50"/>
 
-	<handler-endpoint id="endpoint" input-channel="testChannel" handler="testHandler">
+	<handler-endpoint id="endpoint" input-channel="testChannel"
+	                  handler="testHandler" selector="typeSelector">
 		<schedule period="100"/>
-		<selector ref="typeSelector"/>
 	</handler-endpoint>
 
 	<beans:bean id="typeSelector" class="org.springframework.integration.message.selector.PayloadTypeSelector">

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());