Moved dispatcher policy into channel - simplified MessageBus, DefaultMessageDispatcher, and DispatcherTask.

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

@@ -36,7 +36,6 @@ 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.DispatcherPolicy;
 import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.endpoint.DefaultMessageEndpoint;
@@ -179,22 +178,12 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 	}
 
 	public void registerChannel(String name, MessageChannel channel) {
-		this.registerChannel(name, channel, null);
-	}
-
-	public void registerChannel(String name, MessageChannel channel, DispatcherPolicy dispatcherPolicy) {
 		if (!this.initialized) {
 			this.initialize();
 		}
 		channel.setName(name);
 		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.setMessagingTaskScheduler(this.taskScheduler);
-		if (dispatcherPolicy != null) {
-			dispatcher.setMaxMessagesPerTask(dispatcherPolicy.getMaxMessagesPerTask());
-			dispatcher.setReceiveTimeout(dispatcherPolicy.getReceiveTimeout());
-			dispatcher.setRejectionLimit(dispatcherPolicy.getRejectionLimit());
-			dispatcher.setRetryInterval(dispatcherPolicy.getRetryInterval());
-		}
 		this.dispatchers.put(channel, dispatcher);
 		this.channelRegistry.registerChannel(name, channel);
 		if (logger.isInfoEnabled()) {

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.channel;
 
+import org.springframework.integration.dispatcher.DispatcherPolicy;
 import org.springframework.integration.message.Message;
 
 /**
@@ -36,11 +37,9 @@ public interface MessageChannel {
 	void setName(String name);
 
 	/**
-	 * Return whether this channel has been designated as a publish-subscribe channel.
-	 * If so, any dispatcher retrieving messages from this channel should send each
-	 * message to each of the dispatcher's handlers.
+	 * Return this channel's dispatcher policy
 	 */
-	boolean isPublishSubscribe();
+	DispatcherPolicy getDispatcherPolicy();
 
 	/**
 	 * Send a message, blocking indefinitely if necessary.

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

@@ -21,6 +21,7 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.integration.dispatcher.DispatcherPolicy;
 import org.springframework.integration.message.Message;
 
 /**
@@ -36,23 +37,38 @@ public class SimpleChannel implements MessageChannel, BeanNameAware {
 
 	private String name;
 
-	private boolean publishSubscribe = false;
+	private final DispatcherPolicy dispatcherPolicy;
 
 	private BlockingQueue<Message<?>> queue;
 
 
+	/**
+	 * Create a channel with the specified queue capacity and dispatcher policy.
+	 */
+	public SimpleChannel(int capacity, DispatcherPolicy dispatcherPolicy) {
+		this.queue = new LinkedBlockingQueue<Message<?>>(capacity);
+		this.dispatcherPolicy = (dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy();
+	}
+
 	/**
 	 * Create a channel with the specified queue capacity.
 	 */
 	public SimpleChannel(int capacity) {
-		queue = new LinkedBlockingQueue<Message<?>>(capacity);
+		this(capacity, null);
+	}
+
+	/**
+	 * Create a channel with the default queue capacity and the specified dispatcher policy.
+	 */
+	public SimpleChannel(DispatcherPolicy dispatcherPolicy) {
+		this(DEFAULT_CAPACITY, dispatcherPolicy);
 	}
 
 	/**
 	 * Create a channel with the default queue capacity.
 	 */
 	public SimpleChannel() {
-		this(DEFAULT_CAPACITY);
+		this(DEFAULT_CAPACITY, null);
 	}
 
 
@@ -70,12 +86,8 @@ public class SimpleChannel implements MessageChannel, BeanNameAware {
 		return this.name;
 	}
 
-	public boolean isPublishSubscribe() {
-		return this.publishSubscribe;
-	}
-
-	public void setPublishSubscribe(boolean publishSubscribe) {
-		this.publishSubscribe = publishSubscribe;
+	public DispatcherPolicy getDispatcherPolicy() {
+		return this.dispatcherPolicy;
 	}
 
 	/**

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

@@ -24,6 +24,7 @@ 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.dispatcher.DispatcherPolicy;
 import org.springframework.util.StringUtils;
 
 /**
@@ -39,8 +40,6 @@ public class ChannelParser implements BeanDefinitionParser {
 
 	private static final String PUBLISH_SUBSCRIBE_ATTRIBUTE = "publish-subscribe";
 
-	private static final String PUBLISH_SUBSCRIBE_PROPERTY = "publishSubscribe";
-
 
 	public BeanDefinition parse(Element element, ParserContext parserContext) {
 		RootBeanDefinition channelDef = new RootBeanDefinition(SimpleChannel.class);
@@ -51,7 +50,7 @@ public class ChannelParser implements BeanDefinitionParser {
 		}
 		String isPublishSubscribe = element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE);
 		if ("true".equals(isPublishSubscribe)) {
-			channelDef.getPropertyValues().addPropertyValue(PUBLISH_SUBSCRIBE_PROPERTY, Boolean.TRUE);
+			channelDef.getConstructorArgumentValues().addGenericArgumentValue(new DispatcherPolicy(true));
 		}
 		String beanName = element.getAttribute(ID_ATTRIBUTE);
 		if (!StringUtils.hasText(beanName)) {

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

@@ -27,7 +27,8 @@ import org.springframework.util.Assert;
 /**
  * Message retriever that polls a {@link MessageChannel}. The number of
  * messages retrieved per poll is limited by the '<em>maxMessagesPerTask</em>'
- * property, and the timeout for each receive call is determined by the '<em>receiveTimeout</em>'
+ * property of the channel's dispatcher policy, and the timeout for each receive
+ * call is determined by the dispatcher policy's '<em>receiveTimeout</em>'
  * property. In general, it is recommended to use a value of 1 for
  * 'maxMessagesPerTask' whenever a significant timeout is provided. Otherwise the
  * retriever may be holding on to available messages while waiting for
@@ -39,10 +40,6 @@ public class ChannelPollingMessageRetriever implements MessageRetriever {
 
 	private MessageChannel channel;
 
-	private int maxMessagesPerTask = DispatcherPolicy.DEFAULT_MAX_MESSAGES_PER_TASK;
-
-	private long receiveTimeout = DispatcherPolicy.DEFAULT_RECEIVE_TIMEOUT;
-
 
 	public ChannelPollingMessageRetriever(MessageChannel channel) {
 		Assert.notNull(channel, "'channel' must not be null");
@@ -50,28 +47,20 @@ public class ChannelPollingMessageRetriever implements MessageRetriever {
 	}
 
 
-	public void setMaxMessagesPerTask(int maxMessagesPerTask) {
-		Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least 1");
-		this.maxMessagesPerTask = maxMessagesPerTask;
-	}
-
-	public void setReceiveTimeout(long receiveTimeout) {
-		this.receiveTimeout = receiveTimeout;
-	}
-
 	public MessageChannel getChannel() {
 		return this.channel;
 	}
 
 	public Collection<Message<?>> retrieveMessages() {
 		List<Message<?>> messages = new LinkedList<Message<?>>();
-		while (messages.size() < this.maxMessagesPerTask) {
+		while (messages.size() < this.channel.getDispatcherPolicy().getMaxMessagesPerTask()) {
 			Message<?> message = null;
-			if (this.receiveTimeout < 0) {
+			long timeout = this.channel.getDispatcherPolicy().getReceiveTimeout();
+			if (timeout < 0) {
 				message = this.channel.receive();
 			}
 			else {
-				message = this.channel.receive(this.receiveTimeout);
+				message = this.channel.receive(timeout);
 			}
 			if (message == null) {
 				return messages;

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

@@ -50,16 +50,6 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
 
 	private MessageChannel channel;
 
-	private int maxMessagesPerTask = DispatcherPolicy.DEFAULT_MAX_MESSAGES_PER_TASK;
-
-	private long receiveTimeout = DispatcherPolicy.DEFAULT_RECEIVE_TIMEOUT;
-
-	private int rejectionLimit = DispatcherPolicy.DEFAULT_REJECTION_LIMIT;
-
-	private long retryInterval = DispatcherPolicy.DEFAULT_RETRY_INTERVAL;
-
-	private boolean shouldFailOnRejectionLimit = true;
-
 	private MessagingTaskScheduler scheduler;
 
 	private Schedule defaultSchedule = new PollingSchedule(5);
@@ -79,29 +69,6 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
 	}
 
 
-	public void setMaxMessagesPerTask(int maxMessagesPerTask) {
-		Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least 1");
-		this.maxMessagesPerTask = maxMessagesPerTask;
-	}
-
-	public void setReceiveTimeout(long receiveTimeout) {
-		this.receiveTimeout = receiveTimeout;
-	}
-
-	public void setRejectionLimit(int rejectionLimit) {
-		Assert.isTrue(rejectionLimit > 0, "'rejectionLimit' must be at least 1");
-		this.rejectionLimit = rejectionLimit;
-	}
-
-	public void setShouldFailOnRejectionLimit(boolean shouldFailOnRejectionLimit) {
-		this.shouldFailOnRejectionLimit = shouldFailOnRejectionLimit;
-	}
-
-	public void setRetryInterval(long retryInterval) {
-		Assert.isTrue(retryInterval >= 0, "'retryInterval' must not be negative");
-		this.retryInterval = retryInterval;
-	}
-
 	public void setMessagingTaskScheduler(MessagingTaskScheduler scheduler) {
 		Assert.notNull(scheduler, "'scheduler' must not be null");
 		this.scheduler = scheduler;
@@ -121,7 +88,7 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
 		if (schedule == null) {
 			schedule = this.defaultSchedule;
 		}
-		else if (this.channel.isPublishSubscribe()) {
+		else if (this.channel.getDispatcherPolicy().isPublishSubscribe()) {
 			if (logger.isInfoEnabled()) {
 				logger.info("This dispatcher broadcasts messages for a publish-subscribe channel. " + 
 						"Therefore all handlers are scheduled with its 'defaultSchedule', " +
@@ -163,15 +130,8 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
 			for (Map.Entry<Schedule, List<MessageHandler>> entry : this.scheduledHandlers.entrySet()) {
 				Schedule schedule = entry.getKey();
 				List<MessageHandler> handlers = entry.getValue();
-				ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-				retriever.setMaxMessagesPerTask(this.maxMessagesPerTask);
-				retriever.setReceiveTimeout(this.receiveTimeout);
-				DispatcherTask task = new DispatcherTask(retriever);
+				DispatcherTask task = new DispatcherTask(channel);
 				task.setSchedule(schedule);
-				task.setRejectionLimit(this.rejectionLimit);
-				task.setRetryInterval(this.retryInterval);
-				task.setPublishSubscribe(channel.isPublishSubscribe());
-				task.setShouldFailOnRejectionLimit(this.shouldFailOnRejectionLimit);
 				for (MessageHandler handler : handlers) {
 					if (handler instanceof Lifecycle) {
 						((Lifecycle) handler).start();

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

@@ -34,6 +34,8 @@ public class DispatcherPolicy {
 	public final static long DEFAULT_RETRY_INTERVAL = 1000;
 
 
+	private final boolean publishSubscribe;
+
 	private int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK;
 
 	private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
@@ -42,40 +44,106 @@ public class DispatcherPolicy {
 
 	private long retryInterval = DEFAULT_RETRY_INTERVAL;
 
+	private boolean shouldFailOnRejectionLimit = true;
 
+
+	public DispatcherPolicy() {
+		this.publishSubscribe = false;
+	}
+
+	/**
+	 * Create a DispatcherPolicy.
+	 * 
+	 * @param publishSubscribe whether the dispatcher should attempt to publish
+	 * to all of its subscribed handlers. If '<code>false</code>' it will attempt
+	 * to send to a single handler (point-to-point).
+	 */
+	public DispatcherPolicy(boolean publishSubscribe) {
+		this.publishSubscribe = publishSubscribe;
+	}
+
+
+	/**
+	 * Return whether the dispatcher should attempt to publish to all of its handlers.
+	 * This property is immutable.
+	 */
+	public boolean isPublishSubscribe() {
+		return this.publishSubscribe;
+	}
+
+	/**
+	 * Return the maximum number of messages for each retrieval attempt.
+	 */
 	public int getMaxMessagesPerTask() {
 		return this.maxMessagesPerTask;
 	}
 
+	/**
+	 * Set the maximum number of messages for each retrieval attempt.
+	 */
 	public void setMaxMessagesPerTask(int maxMessagesPerTask) {
 		Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagePerTask' must be at least 1");
 		this.maxMessagesPerTask = maxMessagesPerTask;
 	}
 
+	/**
+	 * Return the maximum amount of time in milliseconds to wait for a message to be available. 
+	 */
 	public long getReceiveTimeout() {
 		return this.receiveTimeout;
 	}
 
+	/**
+	 * Set the maximum amount of time in milliseconds to wait for a message to be available. 
+	 */
 	public void setReceiveTimeout(long receiveTimeout) {
 		this.receiveTimeout = receiveTimeout;
 	}
 
+	/**
+	 * Return the maximum number of retries upon rejection. 
+	 */
 	public int getRejectionLimit() {
 		return this.rejectionLimit;
 	}
 
+	/**
+	 * Set the maximum number of retries upon rejection. 
+	 */
 	public void setRejectionLimit(int rejectionLimit) {
 		Assert.isTrue(rejectionLimit > 0, "'rejectionLimit' must be at least 1");
 		this.rejectionLimit = rejectionLimit;
 	}
 
+	/**
+	 * Return the amount of time in milliseconds to wait between rejections.
+	 */
 	public long getRetryInterval() {
 		return this.retryInterval;
 	}
 
+	/**
+	 * Set the amount of time in milliseconds to wait between rejections.
+	 */
 	public void setRetryInterval(long retryInterval) {
 		Assert.isTrue(retryInterval >= 0, "'retryInterval' must not be negative");
 		this.retryInterval = retryInterval;
 	}
 
+	/**
+	 * Return whether an exception should be thrown when this dispatcher's
+	 * {@link #rejectionLimit} is reached.
+	 */
+	public boolean getShouldFailOnRejectionLimit() {
+		return this.shouldFailOnRejectionLimit;
+	}
+
+	/**
+	 * Specify whether an exception should be thrown when this dispatcher's
+	 * {@link #rejectionLimit} is reached. The default value is 'true'.
+	 */
+	public void setShouldFailOnRejectionLimit(boolean shouldFailOnRejectionLimit) {
+		this.shouldFailOnRejectionLimit = shouldFailOnRejectionLimit;
+	}
+
 }

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

@@ -46,15 +46,9 @@ public class DispatcherTask implements MessagingTask {
 
 	private Log logger = LogFactory.getLog(this.getClass());
 
-	private boolean publishSubscribe = false;
-
 	private Schedule schedule;
 
-	private int rejectionLimit = 5;
-
-	private long retryInterval = 1000;
-
-	private boolean shouldFailOnRejectionLimit = true;
+	private DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
 
 	private MessageRetriever retriever;
 
@@ -64,22 +58,20 @@ public class DispatcherTask implements MessagingTask {
 	public DispatcherTask(MessageChannel channel) {
 		Assert.notNull(channel, "'channel' must not be null");
 		this.retriever = new ChannelPollingMessageRetriever(channel);
-		this.publishSubscribe = channel.isPublishSubscribe();
+		DispatcherPolicy dispatcherPolicy = channel.getDispatcherPolicy();
+		if (dispatcherPolicy != null) {
+			this.dispatcherPolicy = dispatcherPolicy;
+		}
 	}
 
 	public DispatcherTask(MessageRetriever retriever) {
 		Assert.notNull(retriever, "'retriever' must not be null");
 		if (retriever instanceof ChannelPollingMessageRetriever) {
-			this.publishSubscribe = ((ChannelPollingMessageRetriever) retriever).getChannel().isPublishSubscribe();
+			this.dispatcherPolicy = ((ChannelPollingMessageRetriever) retriever).getChannel().getDispatcherPolicy();
 		}
 		this.retriever = retriever;
 	}
 
-
-	public void setPublishSubscribe(boolean publishSubscribe) {
-		this.publishSubscribe = publishSubscribe;
-	}
-
 	public void setSchedule(Schedule schedule) {
 		Assert.notNull(schedule, "'schedule' must not be null");
 		this.schedule = schedule;
@@ -89,24 +81,6 @@ public class DispatcherTask implements MessagingTask {
 		return this.schedule;
 	}
 
-	public void setRejectionLimit(int rejectionLimit) {
-		Assert.isTrue(rejectionLimit > 0, "'rejectionLimit' must be at least 1");
-		this.rejectionLimit = rejectionLimit;
-	}
-
-	public void setRetryInterval(long retryInterval) {
-		Assert.isTrue(retryInterval > 0, "'retryInterval' must not be negative");
-		this.retryInterval = retryInterval;
-	}
-
-	/**
-	 * Specify whether an exception should be thrown when this dispatcher's
-	 * {@link #rejectionLimit} is reached. The default value is 'true'.
-	 */
-	public void setShouldFailOnRejectionLimit(boolean shouldFailOnRejectionLimit) {
-		this.shouldFailOnRejectionLimit = shouldFailOnRejectionLimit;
-	}
-
 	public void addHandler(MessageHandler handler) {
 		Assert.notNull(handler, "'handler' must not be null");
 		this.handlers.add(handler);
@@ -134,15 +108,15 @@ public class DispatcherTask implements MessagingTask {
 	protected boolean dispatchMessage(Message<?> message) {
 		int attempts = 0;
 		List<MessageHandler> targets = new ArrayList<MessageHandler>(this.handlers);
-		while (attempts < this.rejectionLimit) {
+		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 " + this.retryInterval
-							+ " milliseconds");
+							+ " attempt(s), will try again after 'retryInterval' of " +
+							this.dispatcherPolicy.getRetryInterval() + " milliseconds");
 				}
 				try {
-					Thread.sleep(this.retryInterval);
+					Thread.sleep(this.dispatcherPolicy.getRetryInterval());
 				}
 				catch (InterruptedException iex) {
 					Thread.currentThread().interrupt();
@@ -161,7 +135,7 @@ public class DispatcherTask implements MessagingTask {
 				MessageHandler handler = iter.next();
 				try {
 					handler.handle(message);
-					if (!this.publishSubscribe) {
+					if (!this.dispatcherPolicy.isPublishSubscribe()) {
 						return true;
 					}
 					iter.remove();
@@ -188,9 +162,9 @@ public class DispatcherTask implements MessagingTask {
 			}
 			attempts++;
 		}
-		if (this.shouldFailOnRejectionLimit) {
-			throw new MessageDeliveryException("Dispatcher reached rejection limit of " + this.rejectionLimit
-					+ ". Consider increasing the executor's concurrency and/or raising the 'rejectionLimit'.");
+		if (this.dispatcherPolicy.getShouldFailOnRejectionLimit()) {
+			throw new MessageDeliveryException("Dispatcher reached rejection limit of " + this.dispatcherPolicy.getRejectionLimit()
+					+ ". Consider increasing the handler's concurrency and/or the dispatcherPolicy's 'rejectionLimit'.");
 		}
 		return false;
 	}

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

@@ -26,6 +26,7 @@ import org.junit.Test;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.dispatcher.ChannelPollingMessageRetriever;
+import org.springframework.integration.dispatcher.DispatcherPolicy;
 import org.springframework.integration.dispatcher.DispatcherTask;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.StringMessage;
@@ -70,10 +71,11 @@ public class ByteStreamTargetAdapterTests {
 	@Test
 	public void testMaxMessagesPerTaskSameAsMessageCount() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setMaxMessagesPerTask(3);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
 		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-		retriever.setMaxMessagesPerTask(3);
 		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
 		dispatcherTask.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}));
@@ -89,10 +91,11 @@ public class ByteStreamTargetAdapterTests {
 	@Test
 	public void testMaxMessagesPerTaskLessThanMessageCount() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setMaxMessagesPerTask(2);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
 		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-		retriever.setMaxMessagesPerTask(2);
 		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
 		dispatcherTask.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}));
@@ -107,12 +110,12 @@ public class ByteStreamTargetAdapterTests {
 	@Test
 	public void testMaxMessagesPerTaskExceedsMessageCount() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
-		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-		retriever.setMaxMessagesPerTask(5);
-		retriever.setReceiveTimeout(0);
-		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setMaxMessagesPerTask(5);
+		dispatcherPolicy.setReceiveTimeout(0);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		dispatcherTask.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}));
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}));
@@ -126,12 +129,12 @@ public class ByteStreamTargetAdapterTests {
 	@Test
 	public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
-		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-		retriever.setMaxMessagesPerTask(2);
-		retriever.setReceiveTimeout(0);
-		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setMaxMessagesPerTask(2);
+		dispatcherPolicy.setReceiveTimeout(0);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		dispatcherTask.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}));
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}));
@@ -150,12 +153,12 @@ public class ByteStreamTargetAdapterTests {
 	@Test
 	public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
-		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-		retriever.setMaxMessagesPerTask(5);
-		retriever.setReceiveTimeout(0);
-		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setMaxMessagesPerTask(5);
+		dispatcherPolicy.setReceiveTimeout(0);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		dispatcherTask.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}));
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}));
@@ -173,11 +176,12 @@ public class ByteStreamTargetAdapterTests {
 	@Test
 	public void testStreamResetBetweenDispatches() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setMaxMessagesPerTask(2);
+		dispatcherPolicy.setReceiveTimeout(0);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
 		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-		retriever.setMaxMessagesPerTask(2);
-		retriever.setReceiveTimeout(0);
 		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
 		dispatcherTask.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}));
@@ -196,12 +200,12 @@ public class ByteStreamTargetAdapterTests {
 	@Test
 	public void testStreamWriteBetweenDispatches() throws IOException {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream);
-		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-		retriever.setMaxMessagesPerTask(2);
-		retriever.setReceiveTimeout(0);
-		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setMaxMessagesPerTask(2);
+		dispatcherPolicy.setReceiveTimeout(0);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		dispatcherTask.addHandler(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}));
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}));

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

@@ -24,7 +24,7 @@ import org.junit.Test;
 
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
-import org.springframework.integration.dispatcher.ChannelPollingMessageRetriever;
+import org.springframework.integration.dispatcher.DispatcherPolicy;
 import org.springframework.integration.dispatcher.DispatcherTask;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.StringMessage;
@@ -87,11 +87,11 @@ public class CharacterStreamTargetAdapterTests {
 	@Test
 	public void testMaxMessagesPerTaskSameAsMessageCount() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-		retriever.setMaxMessagesPerTask(2);
-		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setMaxMessagesPerTask(2);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		dispatcherTask.addHandler(adapter);
 		channel.send(new StringMessage("foo"));
 		channel.send(new StringMessage("bar"));
@@ -103,13 +103,13 @@ public class CharacterStreamTargetAdapterTests {
 	@Test
 	public void testMaxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setMaxMessagesPerTask(10);
+		dispatcherPolicy.setReceiveTimeout(0);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		adapter.setShouldAppendNewLine(true);
-		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-		retriever.setReceiveTimeout(0);
-		retriever.setMaxMessagesPerTask(10);
-		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
 		dispatcherTask.addHandler(adapter);
 		channel.send(new StringMessage("foo"));
 		channel.send(new StringMessage("bar"));
@@ -137,12 +137,12 @@ public class CharacterStreamTargetAdapterTests {
 	@Test
 	public void testTwoNonStringObjectWithOutNewLines() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
-		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-		retriever.setReceiveTimeout(0);
-		retriever.setMaxMessagesPerTask(2);
-		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setReceiveTimeout(0);
+		dispatcherPolicy.setMaxMessagesPerTask(2);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		dispatcherTask.addHandler(adapter);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");
@@ -156,13 +156,13 @@ public class CharacterStreamTargetAdapterTests {
 	@Test
 	public void testTwoNonStringObjectWithNewLines() {
 		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setReceiveTimeout(0);
+		dispatcherPolicy.setMaxMessagesPerTask(2);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
 		adapter.setShouldAppendNewLine(true);
-		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
-		retriever.setReceiveTimeout(0);
-		retriever.setMaxMessagesPerTask(2);
-		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
 		dispatcherTask.addHandler(adapter);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");

spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java

@@ -33,6 +33,7 @@ import org.springframework.integration.adapter.PollingSourceAdapter;
 import org.springframework.integration.adapter.SourceAdapter;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
+import org.springframework.integration.dispatcher.DispatcherPolicy;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.ErrorMessage;
 import org.springframework.integration.message.GenericMessage;
@@ -127,8 +128,8 @@ public class MessageBusTests {
 
 	@Test
 	public void testBothHandlersReceivePublishSubscribeMessage() {
-		SimpleChannel inputChannel = new SimpleChannel();
-		inputChannel.setPublishSubscribe(true);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(true);
+		SimpleChannel inputChannel = new SimpleChannel(dispatcherPolicy);
 		SimpleChannel outputChannel1 = new SimpleChannel();
 		SimpleChannel outputChannel2 = new SimpleChannel();
 		MessageHandler handler1 = new MessageHandler() {

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

@@ -76,8 +76,7 @@ public class DefaultMessageDispatcherTests {
 		final CountDownLatch latch = new CountDownLatch(2);
 		MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
 		MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
-		SimpleChannel channel = new SimpleChannel();
-		channel.setPublishSubscribe(true);
+		SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true));
 		channel.send(new StringMessage(1, "test"));
 		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1));
@@ -123,8 +122,7 @@ public class DefaultMessageDispatcherTests {
 		MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
 		MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
 		MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
-		SimpleChannel channel = new SimpleChannel();
-		channel.setPublishSubscribe(true);
+		SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true));
 		channel.send(new StringMessage(1, "test"));
 		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1));
@@ -158,12 +156,11 @@ public class DefaultMessageDispatcherTests {
 		MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
 		MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
 		MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
-		SimpleChannel channel = new SimpleChannel();
-		channel.setPublishSubscribe(true);
+		SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true));
+		channel.getDispatcherPolicy().setRejectionLimit(2);
+		channel.getDispatcherPolicy().setRetryInterval(3);
 		channel.send(new StringMessage(1, "test"));
 		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
-		dispatcher.setRejectionLimit(2);
-		dispatcher.setRetryInterval(3);
 		dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1));
 		dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) {
 			@Override
@@ -192,13 +189,12 @@ public class DefaultMessageDispatcherTests {
 		final CountDownLatch latch = new CountDownLatch(3);
 		MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
 		MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
-		SimpleChannel channel = new SimpleChannel();
-		channel.setPublishSubscribe(true);
+		SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true));
+		channel.getDispatcherPolicy().setRejectionLimit(2);
+		channel.getDispatcherPolicy().setRetryInterval(3);
+		channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false);
 		channel.send(new StringMessage(1, "test"));
 		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
-		dispatcher.setRejectionLimit(2);
-		dispatcher.setRetryInterval(3);
-		dispatcher.setShouldFailOnRejectionLimit(false);
 		dispatcher.addHandler(handler1);
 		dispatcher.addHandler(new MessageHandler() {
 			public Message<?> handle(Message<?> message) {
@@ -215,30 +211,16 @@ public class DefaultMessageDispatcherTests {
 
 	@Test
 	public void testNonBroadcastingDispatcherReachesRejectionLimitAndShouldFail() throws InterruptedException {
-		final AtomicInteger counter1 = new AtomicInteger();
-		final AtomicInteger counter2 = new AtomicInteger();
 		final CountDownLatch latch = new CountDownLatch(4);
-		MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
-		MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
-		SimpleChannel channel = new SimpleChannel();
+		MessageHandler handler1 = TestHandlers.rejectingCountDownHandler(latch);
+		MessageHandler handler2 = TestHandlers.rejectingCountDownHandler(latch);
+		SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(false));
 		channel.send(new StringMessage(1, "test"));
 		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
-		dispatcher.setRejectionLimit(2);
-		dispatcher.setRetryInterval(3);
-		dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) {
-			@Override
-			public Message<?> handle(Message<?> message) {
-				latch.countDown();
-				throw new MessageHandlerRejectedExecutionException();
-			}
-		});
-		dispatcher.addHandler(new ConcurrentHandler(handler2, 1, 1) {
-			@Override
-			public Message<?> handle(Message<?> message) {
-				latch.countDown();
-				throw new MessageHandlerRejectedExecutionException();
-			}
-		});
+		channel.getDispatcherPolicy().setRejectionLimit(2);
+		channel.getDispatcherPolicy().setRetryInterval(3);
+		dispatcher.addHandler(handler1);
+		dispatcher.addHandler(handler2);
 		SimpleChannel errorChannel = new SimpleChannel();
 		SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler();
 		scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel));
@@ -262,11 +244,11 @@ public class DefaultMessageDispatcherTests {
 		MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
 		MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
 		SimpleChannel channel = new SimpleChannel();
+		channel.getDispatcherPolicy().setRejectionLimit(2);
+		channel.getDispatcherPolicy().setRetryInterval(3);
+		channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false);
 		channel.send(new StringMessage(1, "test"));
 		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
-		dispatcher.setRejectionLimit(2);
-		dispatcher.setRetryInterval(3);
-		dispatcher.setShouldFailOnRejectionLimit(false);
 		dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) {
 			@Override
 			public Message<?> handle(Message<?> message) {
@@ -304,12 +286,13 @@ public class DefaultMessageDispatcherTests {
 		MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
 		MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
 		MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
-		SimpleChannel channel = new SimpleChannel();
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setRejectionLimit(2);
+		dispatcherPolicy.setRetryInterval(3);
+		dispatcherPolicy.setShouldFailOnRejectionLimit(false);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
 		channel.send(new StringMessage(1, "test"));
 		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
-		dispatcher.setRejectionLimit(2);
-		dispatcher.setRetryInterval(3);
-		dispatcher.setShouldFailOnRejectionLimit(false);
 		dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) {
 			@Override
 			public Message<?> handle(Message<?> message) {
@@ -357,13 +340,13 @@ public class DefaultMessageDispatcherTests {
 		final CountDownLatch latch = new CountDownLatch(8);
 		MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
 		MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
-		SimpleChannel channel = new SimpleChannel();
-		channel.setPublishSubscribe(true);
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(true);
+		dispatcherPolicy.setRejectionLimit(5);
+		dispatcherPolicy.setRetryInterval(3);
+		dispatcherPolicy.setShouldFailOnRejectionLimit(false);
+		SimpleChannel channel = new SimpleChannel(dispatcherPolicy);
 		channel.send(new StringMessage(1, "test"));
 		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
-		dispatcher.setRejectionLimit(5);
-		dispatcher.setRetryInterval(3);
-		dispatcher.setShouldFailOnRejectionLimit(false);
 		dispatcher.addHandler(new ConcurrentHandler(handler1, 1, 1) {
 			@Override
 			public Message<?> handle(Message<?> message) {
@@ -471,8 +454,7 @@ public class DefaultMessageDispatcherTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
 		MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
-		SimpleChannel channel = new SimpleChannel();
-		channel.setPublishSubscribe(true);
+		SimpleChannel channel = new SimpleChannel(new DispatcherPolicy(true));
 		channel.send(new StringMessage(1, "test"));
 		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint();

spring-integration-core/src/test/java/org/springframework/integration/handler/TestHandlers.java

@@ -19,6 +19,7 @@ package org.springframework.integration.handler;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException;
 import org.springframework.integration.message.Message;
 
 /**
@@ -40,6 +41,30 @@ public abstract class TestHandlers {
 		};
 	}
 
+	/**
+	 * Create a {@link MessageHandler} that throws a {@link MessageHandlerRejectedExecutionException}.
+	 */
+	public final static MessageHandler rejectingHandler() {
+		return new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				throw new MessageHandlerRejectedExecutionException();
+			}
+		};
+	}
+
+	/**
+	 * Create a {@link MessageHandler} that counts down on the provided latch and 
+	 * then throws a {@link MessageHandlerRejectedExecutionException}.
+	 */
+	public final static MessageHandler rejectingCountDownHandler(final CountDownLatch latch) {
+		return new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				latch.countDown();
+				throw new MessageHandlerRejectedExecutionException();
+			}
+		};
+	}
+
 	/**
 	 * Create a {@link MessageHandler} that increments the provided counter.
 	 */