Refactored DefaultMessageDispatcher to delegate to MessageDistributor with logic previously in the DispatcherTask. The DispatcherTask is now an inner class of DefaultMessageDispatcher. Also added ChannelPurger (INT-105), MessageSelectingInterceptor (INT-98), and support for "datatype channels" (INT-99).

spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java

@@ -36,7 +36,7 @@ import org.springframework.integration.channel.DefaultChannelRegistry;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
-import org.springframework.integration.dispatcher.MessageDispatcher;
+import org.springframework.integration.dispatcher.SchedulingMessageDispatcher;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.endpoint.DefaultMessageEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
@@ -63,7 +63,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 
 	private Map<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
 
-	private Map<MessageChannel, MessageDispatcher> dispatchers = new ConcurrentHashMap<MessageChannel, MessageDispatcher>();
+	private Map<MessageChannel, SchedulingMessageDispatcher> dispatchers = new ConcurrentHashMap<MessageChannel, SchedulingMessageDispatcher>();
 
 	private List<Lifecycle> lifecycleSourceAdapters = new CopyOnWriteArrayList<Lifecycle>();
 
@@ -194,7 +194,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 	public MessageChannel unregisterChannel(String name) {
 		MessageChannel removedChannel = this.channelRegistry.unregisterChannel(name);
 		if (removedChannel != null) {
-			MessageDispatcher removedDispatcher = this.dispatchers.remove(removedChannel);
+			SchedulingMessageDispatcher removedDispatcher = this.dispatchers.remove(removedChannel);
 			if (removedDispatcher != null && removedDispatcher.isRunning()) {
 				removedDispatcher.stop();
 			}
@@ -300,7 +300,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 	}
 
 	private void registerWithDispatcher(MessageChannel channel, MessageHandler handler, Schedule schedule) {
-		MessageDispatcher dispatcher = dispatchers.get(channel);
+		SchedulingMessageDispatcher dispatcher = dispatchers.get(channel);
 		if (dispatcher == null) {
 			if (logger.isWarnEnabled()) {
 				logger.warn("no dispatcher available for channel '" + channel.getName() + "', be sure to register the channel");
@@ -329,7 +329,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		synchronized (this.lifecycleMonitor) {
 			this.activateEndpoints();
 			this.taskScheduler.start();
-			for (MessageDispatcher dispatcher : this.dispatchers.values()) {
+			for (SchedulingMessageDispatcher dispatcher : this.dispatchers.values()) {
 				dispatcher.start();
 				if (logger.isInfoEnabled()) {
 					logger.info("started dispatcher '" + dispatcher + "'");
@@ -362,7 +362,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 					logger.info("stopped source adapter '" + adapter + "'");
 				}
 			}
-			for (MessageDispatcher dispatcher : this.dispatchers.values()) {
+			for (SchedulingMessageDispatcher dispatcher : this.dispatchers.values()) {
 				dispatcher.stop();
 				if (logger.isInfoEnabled()) {
 					logger.info("stopped dispatcher '" + dispatcher + "'");

spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptor.java

@@ -27,12 +27,12 @@ import org.springframework.integration.message.Message;
  */
 public interface ChannelInterceptor {
 
-	boolean preSend(Message message, MessageChannel channel);
+	boolean preSend(Message<?> message, MessageChannel channel);
 
-	void postSend(Message message, MessageChannel channel, boolean sent);
+	void postSend(Message<?> message, MessageChannel channel, boolean sent);
 
 	boolean preReceive(MessageChannel channel);
 
-	void postReceive(Message message, MessageChannel channel);
+	void postReceive(Message<?> message, MessageChannel channel);
 
 }

spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPurger.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2002-2007 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.channel;
+
+import java.util.List;
+
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.selector.MessageSelector;
+
+/**
+ * A utility class for purging {@link Message Messages} from a
+ * {@link MessageChannel}. Any message that does <em>not</em> match the
+ * provided {@link MessageSelector} will be removed from the channel. If no
+ * {@link MessageSelector} is provided, then <em>all</em> messages will be
+ * cleared from the channel.
+ * <p>
+ * Note that the {@link #purge()} method operates on a snapshot of the messages
+ * within a channel at the time that the method is invoked. It is therefore
+ * possible that new messages will arrive on the channel during the purge
+ * operation and thus will <em>not</em> be removed. Likewise, messages to be
+ * purged may have been removed from the channel while the operation is taking
+ * place. Such messages will not be included in the returned list.
+ * 
+ * @author Mark Fisher
+ */
+public class ChannelPurger {
+
+	private MessageChannel channel;
+
+	private MessageSelector selector;
+
+
+	public ChannelPurger(MessageChannel channel) {
+		this.channel = channel;
+	}
+
+	public ChannelPurger(MessageChannel channel, MessageSelector selector) {
+		this(channel);
+		this.selector = selector;
+	}
+
+
+	public final List<Message<?>> purge() {
+		if (this.selector == null) {
+			return this.channel.clear();
+		}
+		return this.channel.purge(this.selector);
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java

@@ -20,6 +20,7 @@ import java.util.List;
 
 import org.springframework.integration.dispatcher.DispatcherPolicy;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.selector.MessageSelector;
 
 /**
  * Base channel interface defining common behavior for message sending and receiving.
@@ -88,11 +89,11 @@ public interface MessageChannel {
 	/**
 	 * Remove all {@link Message Messages} from this channel.
 	 */
-	List<Message> clear();
+	List<Message<?>> clear();
 
 	/**
-	 * Remove any expired {@link Message Messages} from this channel.
+	 * Remove any {@link Message Messages} that are not accepted by the provided selector.
 	 */
-	List<Message> purge();
+	List<Message<?>> purge(MessageSelector selector);
 
 }

spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java

@@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit;
 
 import org.springframework.integration.dispatcher.DispatcherPolicy;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.selector.MessageSelector;
 import org.springframework.util.Assert;
 
 /**
@@ -111,20 +112,21 @@ public class SimpleChannel extends AbstractMessageChannel {
 		}
 	}
 
-	public List<Message> clear() {
-		List<Message> clearedMessages = new ArrayList<Message>();
+	public List<Message<?>> clear() {
+		List<Message<?>> clearedMessages = new ArrayList<Message<?>>();
 		this.queue.drainTo(clearedMessages);
 		return clearedMessages;
 	}
 
-	public List<Message> purge() {
-		List<Message> purgedMessages = new ArrayList<Message>();
-		// take a snapshot
+	public List<Message<?>> purge(MessageSelector selector) {
+		if (selector == null) {
+			return this.clear();
+		}
+		List<Message<?>> purgedMessages = new ArrayList<Message<?>>();
 		Object[] array = this.queue.toArray();
 		for (Object o : array) {
-			Message message = (Message) o;
-			if (message.isExpired() && this.queue.remove(message)) {
-				// message was still in the queue
+			Message<?> message = (Message<?>) o;
+			if (!selector.accept(message) && this.queue.remove(message)) {
 				purgedMessages.add(message);
 			}
 		}

spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ChannelInterceptorAdapter.java

@@ -14,8 +14,10 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.channel;
+package org.springframework.integration.channel.interceptor;
 
+import org.springframework.integration.channel.ChannelInterceptor;
+import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.message.Message;
 
 /**
@@ -26,18 +28,18 @@ import org.springframework.integration.message.Message;
  */
 public class ChannelInterceptorAdapter implements ChannelInterceptor {
 
-	public boolean preSend(Message message, MessageChannel channel) {
+	public boolean preSend(Message<?> message, MessageChannel channel) {
 		return true;
 	}
 
-	public void postSend(Message message, MessageChannel channel, boolean sent) {
+	public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
 	}
 
 	public boolean preReceive(MessageChannel channel) {
 		return true;
 	}
 
-	public void postReceive(Message message, MessageChannel channel) {
+	public void postReceive(Message<?> message, MessageChannel channel) {
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptor.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2002-2007 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.channel.interceptor;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.selector.MessageSelector;
+
+/**
+ * A {@link org.springframework.integration.channel.ChannelInterceptor} that
+ * delegates to a list of {@link MessageSelector MessageSelectors} to decide
+ * whether a {@link Message} should be accepted on the {@link MessageChannel}.
+ * 
+ * @author Mark Fisher
+ */
+public class MessageSelectingInterceptor extends ChannelInterceptorAdapter {
+
+	private final List<MessageSelector> selectors;
+
+
+	public MessageSelectingInterceptor(MessageSelector... selectors) {
+		this.selectors = Arrays.asList(selectors);
+	}
+
+
+	@Override
+	public boolean preSend(Message<?> message, MessageChannel channel) {
+		for (MessageSelector selector : this.selectors) {
+			if (!selector.accept(message)) {
+				throw new MessageDeliveryException(
+						"selector '" + selector + "' did not accept message '" + message + "'");
+			}
+		}
+		return true;
+	}
+
+}

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

@@ -21,12 +21,18 @@ import org.w3c.dom.Node;
 import org.w3c.dom.NodeList;
 
 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.GenericBeanDefinition;
+import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.beans.factory.xml.BeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.channel.SimpleChannel;
+import org.springframework.integration.channel.interceptor.MessageSelectingInterceptor;
 import org.springframework.integration.dispatcher.DispatcherPolicy;
+import org.springframework.integration.message.selector.MessageSelector;
+import org.springframework.integration.message.selector.PayloadTypeSelector;
 import org.springframework.util.StringUtils;
 
 /**
@@ -44,6 +50,10 @@ public class ChannelParser implements BeanDefinitionParser {
 
 	private static final String DISPATCHER_POLICY_ELEMENT = "dispatcher-policy";
 
+	private static final String DATATYPE_ATTRIBUTE = "datatype";
+
+	private static final String INTERCEPTORS_PROPERTY = "interceptors";
+
 
 	public BeanDefinition parse(Element element, ParserContext parserContext) {
 		RootBeanDefinition channelDef = new RootBeanDefinition(SimpleChannel.class);
@@ -64,6 +74,24 @@ public class ChannelParser implements BeanDefinitionParser {
 		int capacity = (StringUtils.hasText(capAttr)) ? Integer.parseInt(capAttr) : SimpleChannel.DEFAULT_CAPACITY;
 		channelDef.getConstructorArgumentValues().addIndexedArgumentValue(0, capacity);
 		channelDef.getConstructorArgumentValues().addIndexedArgumentValue(1, dispatcherPolicy);
+		ManagedList interceptors = new ManagedList();
+		String datatypeAttr = element.getAttribute(DATATYPE_ATTRIBUTE);
+		if (StringUtils.hasText(datatypeAttr)) {
+			String[] datatypes = StringUtils.commaDelimitedListToStringArray(datatypeAttr);
+			RootBeanDefinition selectorDef = new RootBeanDefinition(PayloadTypeSelector.class);
+			selectorDef.getConstructorArgumentValues().addGenericArgumentValue(datatypes);
+			String selectorBeanName = parserContext.getReaderContext().generateBeanName(selectorDef);
+			BeanComponentDefinition selectorComponent = new BeanComponentDefinition(selectorDef, selectorBeanName);
+			parserContext.registerBeanComponent(selectorComponent);
+			RootBeanDefinition interceptorDef = new RootBeanDefinition(MessageSelectingInterceptor.class);
+			interceptorDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(selectorBeanName));
+			String interceptorBeanName = parserContext.getReaderContext().generateBeanName(interceptorDef);
+			BeanComponentDefinition interceptorComponent = new BeanComponentDefinition(interceptorDef, interceptorBeanName);
+			parserContext.registerBeanComponent(interceptorComponent);
+			interceptors.add(new RuntimeBeanReference(interceptorBeanName));
+		}
+		// TODO: parse interceptor sub-elements
+		channelDef.getPropertyValues().addPropertyValue(INTERCEPTORS_PROPERTY, interceptors);
 		String beanName = element.getAttribute(ID_ATTRIBUTE);
 		parserContext.registerBeanComponent(new BeanComponentDefinition(channelDef, beanName));
 		return channelDef;

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

@@ -53,6 +53,7 @@
 			<xsd:attribute name="id" type="xsd:ID" use="required"/>
 			<xsd:attribute name="capacity" type="xsd:integer"/>
 			<xsd:attribute name="publish-subscribe" type="xsd:boolean" default="false"/>
+			<xsd:attribute name="datatype" type="xsd:string"/>
 		</xsd:complexType>
 	</xsd:element>
 

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java

@@ -16,11 +16,12 @@
 
 package org.springframework.integration.dispatcher;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -28,6 +29,8 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.scheduling.MessagingTask;
 import org.springframework.integration.scheduling.MessagingTaskScheduler;
 import org.springframework.integration.scheduling.MessagingTaskSchedulerAware;
 import org.springframework.integration.scheduling.PollingSchedule;
@@ -44,11 +47,13 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTaskSchedulerAware {
+public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, MessagingTaskSchedulerAware {
 
 	protected Log logger = LogFactory.getLog(this.getClass());
 
-	private MessageChannel channel;
+	private final MessageChannel channel;
+
+	private final MessageRetriever retriever;
 
 	private MessagingTaskScheduler scheduler;
 
@@ -56,7 +61,7 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
 
 	private Map<Schedule, List<MessageHandler>> scheduledHandlers = new ConcurrentHashMap<Schedule, List<MessageHandler>>();
 
-	private List<ScheduledFuture<?>> futures = new CopyOnWriteArrayList<ScheduledFuture<?>>();
+	private AtomicLong totalMessagesProcessed = new AtomicLong();
 
 	private volatile boolean running;
 
@@ -66,6 +71,7 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
 	public DefaultMessageDispatcher(MessageChannel channel) {
 		Assert.notNull(channel, "'channel' must not be null");
 		this.channel = channel;
+		this.retriever = new ChannelPollingMessageRetriever(this.channel);
 	}
 
 
@@ -127,21 +133,14 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
 			return;
 		}
 		synchronized (this.lifecycleMonitor) {
-			for (Map.Entry<Schedule, List<MessageHandler>> entry : this.scheduledHandlers.entrySet()) {
-				Schedule schedule = entry.getKey();
-				List<MessageHandler> handlers = entry.getValue();
-				DispatcherTask task = new DispatcherTask(channel);
-				task.setSchedule(schedule);
+			for (Schedule schedule : this.scheduledHandlers.keySet()) {
+				List<MessageHandler> handlers = this.scheduledHandlers.get(schedule);
 				for (MessageHandler handler : handlers) {
 					if (handler instanceof Lifecycle) {
 						((Lifecycle) handler).start();
 					}
-					task.addHandler(handler);
-				}
-				ScheduledFuture<?> future = this.scheduler.schedule(task);
-				if (future != null) {
-					futures.add(future);
 				}
+				this.scheduler.schedule(new DispatcherTask(schedule));
 			}
 			this.running = true;
 		}
@@ -152,13 +151,10 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
 			return;
 		}
 		synchronized (this.lifecycleMonitor) {
-			for (ScheduledFuture<?> future : this.futures) {
-				future.cancel(true);
-				for (List<MessageHandler> handlerList : scheduledHandlers.values()) {
-					for (MessageHandler handler : handlerList) {
-						if (handler instanceof Lifecycle) {
-							((Lifecycle) handler).stop();
-						}
+			for (List<MessageHandler> handlerList : scheduledHandlers.values()) {
+				for (MessageHandler handler : handlerList) {
+					if (handler instanceof Lifecycle) {
+						((Lifecycle) handler).stop();
 					}
 				}
 			}
@@ -166,4 +162,57 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
 		}
 	}
 
+	public int dispatch() {
+		MessageDistributor distributor = this.getDistributor(this.defaultSchedule);
+		return this.doDispatch(distributor);
+	}
+
+	private int doDispatch(MessageDistributor distributor) {
+		int messagesProcessed = 0;
+		Collection<Message<?>> messages = this.retriever.retrieveMessages();
+		if (messages == null) {
+			return 0;
+		}
+		for (Message<?> message : messages) {
+			if (distributor.distribute(message)) {
+				messagesProcessed++;
+			}
+		}
+		totalMessagesProcessed.addAndGet(messagesProcessed);
+		return messagesProcessed;
+	}
+
+	private MessageDistributor getDistributor(Schedule schedule) {
+		if (schedule == null) {
+			schedule = this.defaultSchedule;
+		}
+		MessageDistributor distributor = new DefaultMessageDistributor(this.channel.getDispatcherPolicy());
+		for (MessageHandler handler : this.scheduledHandlers.get(schedule)) {
+			distributor.addHandler(handler);
+		}
+		return distributor;
+	}
+
+
+	private class DispatcherTask implements MessagingTask {
+
+		private Schedule schedule;
+
+		private MessageDistributor distributor;
+
+
+		public DispatcherTask(Schedule schedule) {
+			this.schedule = (schedule != null) ? schedule : defaultSchedule;
+			this.distributor = getDistributor(this.schedule);
+		}
+
+		public Schedule getSchedule() {
+			return this.schedule;
+		}
+
+		public void run() {
+			doDispatch(this.distributor);
+		}
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java

@@ -17,7 +17,6 @@
 package org.springframework.integration.dispatcher;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
@@ -27,87 +26,44 @@ import org.apache.commons.logging.LogFactory;
 
 import org.springframework.integration.MessageDeliveryException;
 import org.springframework.integration.MessageHandlingException;
-import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.handler.MessageHandlerNotRunningException;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.selector.MessageSelectorRejectedException;
-import org.springframework.integration.scheduling.MessagingTask;
-import org.springframework.integration.scheduling.Schedule;
 import org.springframework.util.Assert;
 
 /**
- * A task for polling {@link MessageDispatcher MessageDispatchers}. If
- * {@link #broadcast} is set to <code>false</code> (the default), each message
- * will be sent to a single {@link MessageHandler}. Otherwise, each
- * retrieved {@link Message} will be sent to all of the handlers.
+ * Default implementation of the {@link MessageDistributor} interface.
  * 
  * @author Mark Fisher
  */
-public class DispatcherTask implements MessagingTask {
+public class DefaultMessageDistributor implements MessageDistributor {
 
-	private Log logger = LogFactory.getLog(this.getClass());
+	private final Log logger = LogFactory.getLog(this.getClass());
 
-	private Schedule schedule;
+	private final List<MessageHandler> handlers = new CopyOnWriteArrayList<MessageHandler>();
 
-	private DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
-
-	private MessageRetriever retriever;
-
-	private List<MessageHandler> handlers = new CopyOnWriteArrayList<MessageHandler>();
+	private final DispatcherPolicy dispatcherPolicy;
 
 
-	public DispatcherTask(MessageChannel channel) {
-		Assert.notNull(channel, "'channel' must not be null");
-		this.retriever = new ChannelPollingMessageRetriever(channel);
-		DispatcherPolicy dispatcherPolicy = channel.getDispatcherPolicy();
-		if (dispatcherPolicy != null) {
-			this.dispatcherPolicy = dispatcherPolicy;
-		}
+	public DefaultMessageDistributor(DispatcherPolicy dispatcherPolicy) {
+		Assert.notNull(dispatcherPolicy, "'dispatcherPolicy' must not be null");
+		this.dispatcherPolicy = dispatcherPolicy;
 	}
 
 
-	public void setSchedule(Schedule schedule) {
-		Assert.notNull(schedule, "'schedule' must not be null");
-		this.schedule = schedule;
-	}
-
-	public Schedule getSchedule() {
-		return this.schedule;
-	}
-
 	public void addHandler(MessageHandler handler) {
-		Assert.notNull(handler, "'handler' must not be null");
 		this.handlers.add(handler);
 	}
 
-	/**
-	 * Retrieves messages and dispatches to the executors.
-	 * 
-	 * @return the number of messages processed
-	 */
-	public int dispatch() {
-		int messagesProcessed = 0;
-		Collection<Message<?>> messages = this.retriever.retrieveMessages();
-		if (messages == null) {
-			return 0;
-		}
-		for (Message<?> message : messages) {
-			if (dispatchMessage(message)) {
-				messagesProcessed++;
-			}
-		}
-		return messagesProcessed;
-	}
-
-	protected boolean dispatchMessage(Message<?> message) {
+	public boolean distribute(Message<?> message) {
 		int attempts = 0;
 		List<MessageHandler> targets = new ArrayList<MessageHandler>(this.handlers);
 		while (attempts < this.dispatcherPolicy.getRejectionLimit()) {
 			if (attempts > 0) {
 				if (logger.isDebugEnabled()) {
-					logger.debug("handler(s) rejected message after " + attempts
-							+ " attempt(s), will try again after 'retryInterval' of " +
+					logger.debug("handler(s) rejected message after " + attempts +
+							" attempt(s), will try again after 'retryInterval' of " +
 							this.dispatcherPolicy.getRetryInterval() + " milliseconds");
 				}
 				try {
@@ -164,8 +120,4 @@ public class DispatcherTask implements MessagingTask {
 		return false;
 	}
 
-	public void run() {
-		this.dispatch();
-	}
-
 }

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java

@@ -16,19 +16,17 @@
 
 package org.springframework.integration.dispatcher;
 
-import org.springframework.context.Lifecycle;
 import org.springframework.integration.handler.MessageHandler;
-import org.springframework.integration.scheduling.Schedule;
 
 /**
  * Strategy interface for dispatching messages.
  * 
  * @author Mark Fisher
  */
-public interface MessageDispatcher extends Lifecycle {
+public interface MessageDispatcher {
 
 	void addHandler(MessageHandler handler);
 
-	void addHandler(MessageHandler handler, Schedule schedule);
+	int dispatch();
 
 }

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDistributor.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2002-2007 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.dispatcher;
+
+import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.message.Message;
+
+/**
+ * Strategy interface for distributing a {@link Message} to one or more
+ * {@link MessageHandler MessageHandlers}.
+ * 
+ * @author Mark Fisher
+ */
+public interface MessageDistributor {
+
+	void addHandler(MessageHandler handler);
+
+	boolean distribute(Message<?> message);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SchedulingMessageDispatcher.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2002-2007 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.dispatcher;
+
+import org.springframework.context.Lifecycle;
+import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.scheduling.Schedule;
+
+/**
+ * An extension to the {@link MessageDispatcher} strategy for handlers that may
+ * be scheduled.
+ * 
+ * @author Mark Fisher
+ */
+public interface SchedulingMessageDispatcher extends MessageDispatcher, Lifecycle {
+
+	void setDefaultSchedule(Schedule defaultSchedule);
+
+	void addHandler(MessageHandler handler, Schedule schedule);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/message/selector/UnexpiredMessageSelector.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2002-2007 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 org.springframework.integration.message.Message;
+
+/**
+ * A {@link MessageSelector} that accepts {@link Message Messages} that are
+ * <em>not</em> expired.
+ * 
+ * @author Mark Fisher
+ */
+public class UnexpiredMessageSelector implements MessageSelector {
+
+	public boolean accept(Message<?> message) {
+		return (!message.isExpired());
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java

@@ -23,10 +23,10 @@ import java.io.IOException;
 
 import org.junit.Test;
 
-import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
+import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
 import org.springframework.integration.dispatcher.DispatcherPolicy;
-import org.springframework.integration.dispatcher.DispatcherTask;
+import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.StringMessage;
 
@@ -38,13 +38,8 @@ public class ByteStreamTargetAdapterTests {
 	@Test
 	public void testSingleByteArray() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
-		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}));
-		int count = dispatcherTask.dispatch();
-		assertEquals(1, count);
+		adapter.handle(new GenericMessage<byte[]>(new byte[] {1,2,3}));
 		byte[] result = stream.toByteArray();
 		assertEquals(3, result.length);
 		assertEquals(1, result[0]);
@@ -55,13 +50,8 @@ public class ByteStreamTargetAdapterTests {
 	@Test
 	public void testSingleString() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
-		channel.send(new StringMessage("foo"));
-		int count = dispatcherTask.dispatch();
-		assertEquals(1, count);
+		adapter.handle(new StringMessage("foo"));
 		byte[] result = stream.toByteArray();
 		assertEquals(3, result.length);
 		assertEquals("foo", new String(result));
@@ -74,12 +64,12 @@ public class ByteStreamTargetAdapterTests {
 		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
 		dispatcherPolicy.setMaxMessagesPerTask(3);
 		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		assertEquals(3, dispatcherTask.dispatch());
+		assertEquals(3, dispatcher.dispatch());
 		byte[] result = stream.toByteArray();
 		assertEquals(9, result.length);
 		assertEquals(1, result[0]);
@@ -93,12 +83,12 @@ public class ByteStreamTargetAdapterTests {
 		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		assertEquals(2, dispatcherTask.dispatch());
+		assertEquals(2, dispatcher.dispatch());
 		byte[] result = stream.toByteArray();
 		assertEquals(6, result.length);
 		assertEquals(1, result[0]);
@@ -112,12 +102,12 @@ public class ByteStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(5);
 		dispatcherPolicy.setReceiveTimeout(0);
 		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		assertEquals(3, dispatcherTask.dispatch());
+		assertEquals(3, dispatcher.dispatch());
 		byte[] result = stream.toByteArray();
 		assertEquals(9, result.length);
 		assertEquals(1, result[0]);
@@ -131,16 +121,16 @@ public class ByteStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		dispatcherPolicy.setReceiveTimeout(0);
 		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		assertEquals(2, dispatcherTask.dispatch());
+		assertEquals(2, dispatcher.dispatch());
 		byte[] result1 = stream.toByteArray();
 		assertEquals(6, result1.length);
 		assertEquals(1, result1[0]);
-		assertEquals(1, dispatcherTask.dispatch());
+		assertEquals(1, dispatcher.dispatch());
 		byte[] result2 = stream.toByteArray();
 		assertEquals(9, result2.length);
 		assertEquals(1, result2[0]);
@@ -155,16 +145,16 @@ public class ByteStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(5);
 		dispatcherPolicy.setReceiveTimeout(0);
 		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		assertEquals(3, dispatcherTask.dispatch());
+		assertEquals(3, dispatcher.dispatch());
 		byte[] result1 = stream.toByteArray();
 		assertEquals(9, result1.length);
 		assertEquals(1, result1[0]);
-		assertEquals(0, dispatcherTask.dispatch());
+		assertEquals(0, dispatcher.dispatch());
 		byte[] result2 = stream.toByteArray();
 		assertEquals(9, result2.length);	
 		assertEquals(1, result2[0]);
@@ -178,16 +168,16 @@ public class ByteStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		dispatcherPolicy.setReceiveTimeout(0);
 		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		assertEquals(2, dispatcherTask.dispatch());
+		assertEquals(2, dispatcher.dispatch());
 		byte[] result1 = stream.toByteArray();
 		assertEquals(6, result1.length);
 		stream.reset();
-		assertEquals(1, dispatcherTask.dispatch());
+		assertEquals(1, dispatcher.dispatch());
 		byte[] result2 = stream.toByteArray();
 		assertEquals(3, result2.length);
 		assertEquals(7, result2[0]);
@@ -201,17 +191,17 @@ public class ByteStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		dispatcherPolicy.setReceiveTimeout(0);
 		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		assertEquals(2, dispatcherTask.dispatch());
+		assertEquals(2, dispatcher.dispatch());
 		byte[] result1 = stream.toByteArray();
 		assertEquals(6, result1.length);
 		stream.write(new byte[] {123});
 		stream.flush();
-		assertEquals(1, dispatcherTask.dispatch());
+		assertEquals(1, dispatcher.dispatch());
 		byte[] result2 = stream.toByteArray();
 		assertEquals(10, result2.length);
 		assertEquals(1, result2[0]);

spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java

@@ -24,8 +24,9 @@ import org.junit.Test;
 
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
+import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
 import org.springframework.integration.dispatcher.DispatcherPolicy;
-import org.springframework.integration.dispatcher.DispatcherTask;
+import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.StringMessage;
 
@@ -37,13 +38,8 @@ public class CharacterStreamTargetAdapterTests {
 	@Test
 	public void testSingleString() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
-		channel.send(new StringMessage("foo"));
-		int count = dispatcherTask.dispatch();
-		assertEquals(1, count);
+		adapter.handle(new StringMessage("foo"));
 		String result = new String(stream.toByteArray());
 		assertEquals("foo", result);
 	}
@@ -53,14 +49,14 @@ public class CharacterStreamTargetAdapterTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		assertEquals(1, dispatcherTask.dispatch());
+		assertEquals(1, dispatcher.dispatch());
 		String result1 = new String(stream.toByteArray());
 		assertEquals("foo", result1);
-		assertEquals(1, dispatcherTask.dispatch());
+		assertEquals(1, dispatcher.dispatch());
 		String result2 = new String(stream.toByteArray());
 		assertEquals("foobar", result2);
 	}
@@ -71,15 +67,15 @@ public class CharacterStreamTargetAdapterTests {
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
 		adapter.setShouldAppendNewLine(true);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		assertEquals(1, dispatcherTask.dispatch());
+		assertEquals(1, dispatcher.dispatch());
 		String result1 = new String(stream.toByteArray());
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine, result1);
-		assertEquals(1, dispatcherTask.dispatch());
+		assertEquals(1, dispatcher.dispatch());
 		String result2 = new String(stream.toByteArray());
 		assertEquals("foo" + newLine + "bar" + newLine, result2);
 	}
@@ -91,11 +87,11 @@ public class CharacterStreamTargetAdapterTests {
 		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		assertEquals(2, dispatcherTask.dispatch());
+		assertEquals(2, dispatcher.dispatch());
 		String result = new String(stream.toByteArray());
 		assertEquals("foobar", result);
 	}
@@ -108,12 +104,12 @@ public class CharacterStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(10);
 		dispatcherPolicy.setReceiveTimeout(0);
 		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		adapter.setShouldAppendNewLine(true);
-		dispatcherTask.addHandler(adapter);
+		dispatcher.addHandler(adapter);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		assertEquals(2, dispatcherTask.dispatch());
+		assertEquals(2, dispatcher.dispatch());
 		String result = new String(stream.toByteArray());
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine + "bar" + newLine, result);
@@ -124,11 +120,11 @@ public class CharacterStreamTargetAdapterTests {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		TestObject testObject = new TestObject("foo");
 		channel.send(new GenericMessage<TestObject>(testObject));
-		int count = dispatcherTask.dispatch();
+		int count = dispatcher.dispatch();
 		assertEquals(1, count);
 		String result = new String(stream.toByteArray());
 		assertEquals("foo", result);
@@ -142,13 +138,13 @@ public class CharacterStreamTargetAdapterTests {
 		dispatcherPolicy.setReceiveTimeout(0);
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
-		dispatcherTask.addHandler(adapter);
+		MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.addHandler(adapter);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");
 		channel.send(new GenericMessage<TestObject>(testObject1), 0);
 		channel.send(new GenericMessage<TestObject>(testObject2), 0);
-		assertEquals(2, dispatcherTask.dispatch());
+		assertEquals(2, dispatcher.dispatch());
 		String result = new String(stream.toByteArray());
 		assertEquals("foobar", result);
 	}
@@ -161,14 +157,14 @@ public class CharacterStreamTargetAdapterTests {
 		dispatcherPolicy.setReceiveTimeout(0);
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
-		DispatcherTask dispatcherTask = new DispatcherTask(channel);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		adapter.setShouldAppendNewLine(true);
-		dispatcherTask.addHandler(adapter);
+		dispatcher.addHandler(adapter);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");
 		channel.send(new GenericMessage<TestObject>(testObject1), 0);
 		channel.send(new GenericMessage<TestObject>(testObject2), 0);
-		assertEquals(2, dispatcherTask.dispatch());
+		dispatcher.dispatch();
 		String result = new String(stream.toByteArray());
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine + "bar" + newLine, result);

spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2002-2007 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.channel;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import java.util.List;
+
+import org.junit.Test;
+
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.integration.message.selector.MessageSelector;
+
+/**
+ * @author Mark Fisher
+ */
+public class ChannelPurgerTests {
+
+	@Test
+	public void testPurgeAllWithoutSelector() {
+		MessageChannel channel = new SimpleChannel();
+		channel.send(new StringMessage("test1"));
+		channel.send(new StringMessage("test2"));
+		channel.send(new StringMessage("test3"));
+		ChannelPurger purger = new ChannelPurger(channel);
+		List<Message<?>> purgedMessages = purger.purge();
+		assertEquals(3, purgedMessages.size());
+		assertNull(channel.receive(0));
+	}
+
+	@Test
+	public void testPurgeAllWithSelector() {
+		MessageChannel channel = new SimpleChannel();
+		channel.send(new StringMessage("test1"));
+		channel.send(new StringMessage("test2"));
+		channel.send(new StringMessage("test3"));
+		ChannelPurger purger = new ChannelPurger(channel, new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				return false;
+			}
+		});
+		List<Message<?>> purgedMessages = purger.purge();
+		assertEquals(3, purgedMessages.size());
+		assertNull(channel.receive(0));
+	}
+
+	@Test
+	public void testPurgeNoneWithSelector() {
+		MessageChannel channel = new SimpleChannel();
+		channel.send(new StringMessage("test1"));
+		channel.send(new StringMessage("test2"));
+		channel.send(new StringMessage("test3"));
+		ChannelPurger purger = new ChannelPurger(channel, new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				return true;
+			}
+		});
+		List<Message<?>> purgedMessages = purger.purge();
+		assertEquals(0, purgedMessages.size());
+		assertNotNull(channel.receive(0));
+		assertNotNull(channel.receive(0));
+		assertNotNull(channel.receive(0));
+	}
+
+	@Test
+	public void testPurgeSubsetWithSelector() {
+		MessageChannel channel = new SimpleChannel();
+		channel.send(new StringMessage("test1"));
+		channel.send(new StringMessage("test2"));
+		channel.send(new StringMessage("test3"));
+		ChannelPurger purger = new ChannelPurger(channel, new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				return (message.getPayload().equals("test2"));
+			}
+		});
+		List<Message<?>> purgedMessages = purger.purge();
+		assertEquals(2, purgedMessages.size());
+		Message<?> message = channel.receive(0);
+		assertNotNull(message);
+		assertEquals("test2", message.getPayload());
+		assertNull(channel.receive(0));
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java

@@ -34,6 +34,7 @@ import org.junit.Test;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.StringMessage;
+import org.springframework.integration.message.selector.UnexpiredMessageSelector;
 
 /**
  * @author Mark Fisher
@@ -203,7 +204,7 @@ public class SimpleChannelTests {
 		assertTrue(channel.send(message1));
 		assertTrue(channel.send(message2));
 		assertFalse(channel.send(message3, 0));
-		List<Message> clearedMessages = channel.clear();
+		List<Message<?>> clearedMessages = channel.clear();
 		assertNotNull(clearedMessages);
 		assertEquals(2, clearedMessages.size());
 		assertTrue(channel.send(message3));
@@ -212,7 +213,7 @@ public class SimpleChannelTests {
 	@Test
 	public void testClearEmptyChannel() {
 		SimpleChannel channel = new SimpleChannel();
-		List<Message> clearedMessages = channel.clear();
+		List<Message<?>> clearedMessages = channel.clear();
 		assertNotNull(clearedMessages);
 		assertEquals(0, clearedMessages.size());
 	}
@@ -231,7 +232,7 @@ public class SimpleChannelTests {
 		assertTrue(channel.send(expiredMessage, 0));
 		assertTrue(channel.send(unexpiredMessage, 0));
 		assertFalse(channel.send(new StringMessage("atCapacity"), 0));
-		List<Message> purgedMessages = channel.purge();
+		List<Message<?>> purgedMessages = channel.purge(new UnexpiredMessageSelector());
 		assertNotNull(purgedMessages);
 		assertEquals(1, purgedMessages.size());
 		assertTrue(channel.send(new StringMessage("roomAvailable"), 0));

spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java

@@ -14,10 +14,9 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.channel;
+package org.springframework.integration.channel.interceptor;
 
 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.assertSame;
@@ -28,6 +27,8 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.StringMessage;
 

spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptorTests.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2002-2007 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.channel.interceptor;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.channel.SimpleChannel;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.integration.message.selector.MessageSelector;
+
+/**
+ * @author Mark Fisher
+ */
+public class MessageSelectingInterceptorTests {
+
+	@Test
+	public void testSingleSelectorAccepts() {
+		final AtomicInteger counter = new AtomicInteger();
+		MessageSelector selector = new TestMessageSelector(true, counter);
+		MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector);
+		SimpleChannel channel = new SimpleChannel();
+		channel.addInterceptor(interceptor);
+		assertTrue(channel.send(new StringMessage("test1")));
+	}
+
+	@Test(expected=MessageDeliveryException.class)
+	public void testSingleSelectorRejects() {
+		final AtomicInteger counter = new AtomicInteger();
+		MessageSelector selector = new TestMessageSelector(false, counter);
+		MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector);
+		SimpleChannel channel = new SimpleChannel();
+		channel.addInterceptor(interceptor);
+		channel.send(new StringMessage("test1"));
+	}
+
+	@Test
+	public void testMultipleSelectorsAccept() {
+		final AtomicInteger counter = new AtomicInteger();
+		MessageSelector selector1 = new TestMessageSelector(true, counter);
+		MessageSelector selector2 = new TestMessageSelector(true, counter);
+		MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector1, selector2);
+		SimpleChannel channel = new SimpleChannel();
+		channel.addInterceptor(interceptor);
+		assertTrue(channel.send(new StringMessage("test1")));
+		assertEquals(2, counter.get());
+	}
+
+	@Test
+	public void testMultipleSelectorsReject() {
+		boolean exceptionThrown = false;
+		final AtomicInteger counter = new AtomicInteger();
+		MessageSelector selector1 = new TestMessageSelector(true, counter);
+		MessageSelector selector2 = new TestMessageSelector(false, counter);
+		MessageSelector selector3 = new TestMessageSelector(false, counter);
+		MessageSelector selector4 = new TestMessageSelector(true, counter);
+		MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector1, selector2, selector3, selector4);
+		SimpleChannel channel = new SimpleChannel();
+		channel.addInterceptor(interceptor);
+		try {
+			channel.send(new StringMessage("test1"));
+		}
+		catch (MessageDeliveryException e) {
+			exceptionThrown = true;
+		}
+		assertTrue(exceptionThrown);
+		assertEquals(2, counter.get());
+	}
+
+
+	private static class TestMessageSelector implements MessageSelector {
+
+		private final boolean shouldAccept;
+
+		private final AtomicInteger counter;
+
+
+		public TestMessageSelector(boolean shouldAccept, AtomicInteger counter) {
+			this.shouldAccept = shouldAccept;
+			this.counter = counter;
+		}
+
+
+		public boolean accept(Message<?> message) {
+			this.counter.incrementAndGet();
+			return this.shouldAccept;
+		}
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java

@@ -28,6 +28,7 @@ import org.junit.Test;
 
 import org.springframework.beans.FatalBeanException;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.MessageDeliveryException;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
 import org.springframework.integration.dispatcher.DispatcherPolicy;
@@ -143,6 +144,48 @@ public class ChannelParserTests {
 		assertFalse(dispatcherPolicy.getShouldFailOnRejectionLimit());
 	}
 
+	@Test
+	public void testDatatypeChannelWithCorrectType() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"channelParserTests.xml", this.getClass());
+		MessageChannel channel = (MessageChannel) context.getBean("integerChannel");
+		assertTrue(channel.send(new GenericMessage<Integer>(123)));
+	}
+
+	@Test(expected=MessageDeliveryException.class)
+	public void testDatatypeChannelWithIncorrectType() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"channelParserTests.xml", this.getClass());
+		MessageChannel channel = (MessageChannel) context.getBean("integerChannel");
+		channel.send(new StringMessage("incorrect type"));
+	}
+
+	@Test
+	public void testDatatypeChannelWithAssignableSubTypes() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"channelParserTests.xml", this.getClass());
+		MessageChannel channel = (MessageChannel) context.getBean("numberChannel");
+		assertTrue(channel.send(new GenericMessage<Integer>(123)));
+		assertTrue(channel.send(new GenericMessage<Double>(123.45)));
+	}
+
+	@Test
+	public void testMultipleDatatypeChannelWithCorrectTypes() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"channelParserTests.xml", this.getClass());
+		MessageChannel channel = (MessageChannel) context.getBean("stringOrNumberChannel");
+		assertTrue(channel.send(new GenericMessage<Integer>(123)));
+		assertTrue(channel.send(new StringMessage("accepted type")));
+	}
+
+	@Test(expected=MessageDeliveryException.class)
+	public void testMultipleDatatypeChannelWithIncorrectType() {
+		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
+				"channelParserTests.xml", this.getClass());
+		MessageChannel channel = (MessageChannel) context.getBean("stringOrNumberChannel");
+		channel.send(new GenericMessage<Boolean>(true));
+	}
+
 
 	private static class TestHandler implements MessageHandler {
 

spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml

@@ -23,4 +23,10 @@
 		                   should-fail-on-rejection-limit="false"/>
 	</channel>
 
+	<channel id="integerChannel" datatype="java.lang.Integer"/>
+
+	<channel id="numberChannel" datatype="java.lang.Number"/>
+
+	<channel id="stringOrNumberChannel" datatype="java.lang.String,java.lang.Number"/>
+
 </beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDistributorTests.java

@@ -24,52 +24,47 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 
-import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.handler.TestHandlers;
 import org.springframework.integration.message.StringMessage;
 
 /**
  * @author Mark Fisher
  */
-public class DispatcherTaskTests {
+public class DefaultMessageDistributorTests {
 
 	@Test
-	public void testSimpleDispatch() throws InterruptedException {
-		MessageChannel channel = new SimpleChannel();
-		DispatcherTask task = new DispatcherTask(channel);
+	public void testSingleMessage() throws InterruptedException {
+		MessageDistributor distributor = new DefaultMessageDistributor(new DispatcherPolicy());
 		final CountDownLatch latch = new CountDownLatch(1);
-		task.addHandler(TestHandlers.countDownHandler(latch));
-		task.dispatchMessage(new StringMessage("test"));
+		distributor.addHandler(TestHandlers.countDownHandler(latch));
+		distributor.distribute(new StringMessage("test"));
 		latch.await(500, TimeUnit.MILLISECONDS);
 		assertEquals(0, latch.getCount());
 	}
 
 	@Test
-	public void testDispatchWithPointToPointChannel() throws InterruptedException {
-		MessageChannel channel = new SimpleChannel(new DispatcherPolicy(false));
-		DispatcherTask task = new DispatcherTask(channel);
+	public void testPointToPoint() throws InterruptedException {
+		MessageDistributor distributor = new DefaultMessageDistributor(new DispatcherPolicy(false));
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicInteger counter1 = new AtomicInteger();
 		final AtomicInteger counter2 = new AtomicInteger();
-		task.addHandler(TestHandlers.countingCountDownHandler(counter1, latch));
-		task.addHandler(TestHandlers.countingCountDownHandler(counter2, latch));
-		task.dispatchMessage(new StringMessage("test"));
+		distributor.addHandler(TestHandlers.countingCountDownHandler(counter1, latch));
+		distributor.addHandler(TestHandlers.countingCountDownHandler(counter2, latch));
+		distributor.distribute(new StringMessage("test"));
 		latch.await(500, TimeUnit.MILLISECONDS);
 		assertEquals(0, latch.getCount());
 		assertEquals("only 1 handler should have received the message", 1, counter1.get() + counter2.get());
 	}
 
 	@Test
-	public void testDispatchWithPublishSubscribeChannel() throws InterruptedException {
-		MessageChannel channel = new SimpleChannel(new DispatcherPolicy(true));
-		DispatcherTask task = new DispatcherTask(channel);
+	public void testPublishSubscribe() throws InterruptedException {
+		MessageDistributor distributor = new DefaultMessageDistributor(new DispatcherPolicy(true));
 		final CountDownLatch latch = new CountDownLatch(2);
 		final AtomicInteger counter1 = new AtomicInteger();
 		final AtomicInteger counter2 = new AtomicInteger();
-		task.addHandler(TestHandlers.countingCountDownHandler(counter1, latch));
-		task.addHandler(TestHandlers.countingCountDownHandler(counter2, latch));
-		task.dispatchMessage(new StringMessage("test"));
+		distributor.addHandler(TestHandlers.countingCountDownHandler(counter1, latch));
+		distributor.addHandler(TestHandlers.countingCountDownHandler(counter2, latch));
+		distributor.distribute(new StringMessage("test"));
 		latch.await(500, TimeUnit.MILLISECONDS);
 		assertEquals(0, latch.getCount());
 		assertEquals(1, counter1.get());

spring-integration-core/src/test/java/org/springframework/integration/message/selector/UnexpiredMessageSelectorTests.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2002-2007 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 static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Date;
+
+import org.junit.Test;
+
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.StringMessage;
+
+/**
+ * @author Mark Fisher
+ */
+public class UnexpiredMessageSelectorTests {
+
+	@Test
+	public void testExpiredMessageRejected() {
+		long past = System.currentTimeMillis() - 60000;
+		Message<?> message = new StringMessage("expired");
+		message.getHeader().setExpiration(new Date(past));
+		UnexpiredMessageSelector selector = new UnexpiredMessageSelector();
+		assertFalse(selector.accept(message));
+	}
+
+	@Test
+	public void testUnexpiredMessageAccepted() {
+		long future = System.currentTimeMillis() + 60000;
+		Message<?> message = new StringMessage("unexpired");
+		message.getHeader().setExpiration(new Date(future));
+		UnexpiredMessageSelector selector = new UnexpiredMessageSelector();
+		assertTrue(selector.accept(message));
+	}
+
+}