DefaultMessageEndpoint now provides a ReplyHandler callback for ConcurrentHandlers.

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

@@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.integration.dispatcher.DispatcherPolicy;
 import org.springframework.integration.message.Message;
+import org.springframework.util.Assert;
 
 /**
  * Simple implementation of a message channel. Each {@link Message} is
@@ -33,7 +34,7 @@ import org.springframework.integration.message.Message;
  */
 public class SimpleChannel implements MessageChannel, BeanNameAware {
 
-	private static final int DEFAULT_CAPACITY = 25;
+	private static final int DEFAULT_CAPACITY = 100;
 
 	private String name;
 
@@ -110,6 +111,7 @@ public class SimpleChannel implements MessageChannel, BeanNameAware {
 	 * <code>false</code> if the sending thread is interrupted.
 	 */
 	public boolean send(Message message) {
+		Assert.notNull(message, "'message' must not be null");
 		try {
 			queue.put(message);
 			return true;
@@ -134,6 +136,7 @@ public class SimpleChannel implements MessageChannel, BeanNameAware {
 	 * time or the sending thread is interrupted.
 	 */
 	public boolean send(Message message, long timeout) {
+		Assert.notNull(message, "'message' must not be null");
 		try {
 			if (timeout > 0) {
 				return this.queue.offer(message, timeout, TimeUnit.MILLISECONDS);

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

@@ -29,7 +29,9 @@ 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;

spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java

@@ -28,15 +28,18 @@ import org.springframework.integration.bus.Subscription;
 import org.springframework.integration.channel.ChannelRegistry;
 import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.dispatcher.MessageHandlerNotRunningException;
-import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException;
-import org.springframework.integration.dispatcher.MessageSelectorRejectedException;
 import org.springframework.integration.handler.ConcurrentHandler;
 import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.handler.MessageHandlerNotRunningException;
+import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
+import org.springframework.integration.handler.ReplyHandler;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageHeader;
 import org.springframework.integration.message.selector.MessageSelector;
+import org.springframework.integration.message.selector.MessageSelectorRejectedException;
 import org.springframework.integration.util.ErrorHandler;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * Default implementation of the {@link MessageEndpoint} interface.
@@ -57,6 +60,8 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 
 	private ErrorHandler errorHandler;
 
+	private ReplyHandler replyHandler = new EndpointReplyHandler();
+
 	private String defaultOutputChannelName;
 
 	private ChannelRegistry channelRegistry;
@@ -145,7 +150,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 	}
 
 	public void afterPropertiesSet() {
-		if (this.concurrencyPolicy != null) {
+		if (this.concurrencyPolicy != null || this.handler instanceof ConcurrentHandler) {
 			if (!(this.handler instanceof ConcurrentHandler)) {
 				this.handler = new ConcurrentHandler(this.handler, this.concurrencyPolicy);
 			}
@@ -153,6 +158,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 			if (this.errorHandler != null) {
 				concurrentHandler.setErrorHandler(this.errorHandler);
 			}
+			concurrentHandler.setReplyHandler(this.replyHandler);
 			concurrentHandler.afterPropertiesSet();
 		}
 		this.initialized = true;
@@ -206,13 +212,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 		try {
 			Message<?> replyMessage = handler.handle(message);
 			if (replyMessage != null) {
-				MessageChannel replyChannel = this.resolveReplyChannel(message);
-				if (replyChannel == null) {
-					throw new MessageHandlingException("Unable to determine reply channel for message. "
-							+ "Provide a 'replyChannelName' in the message header or a 'defaultOutputChannelName' "
-							+ "on the message endpoint.");
-				}
-				replyChannel.send(replyMessage);
+				this.replyHandler.handle(replyMessage, message.getHeader());
 			}
 		}
 		catch (MessageHandlerRejectedExecutionException e) {
@@ -228,12 +228,11 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 		return null;
 	}
 
-	private MessageChannel resolveReplyChannel(Message<?> message) {
+	private MessageChannel resolveReplyChannel(String replyChannelName) {
 		if (this.channelRegistry == null) {
 			return null;
 		}
-		String replyChannelName = message.getHeader().getReplyChannelName();
-		if (replyChannelName != null && replyChannelName.trim().length() > 0) {
+		if (StringUtils.hasText(replyChannelName)) {
 			return this.channelRegistry.lookupChannel(replyChannelName);
 		}
 		if (this.defaultOutputChannelName != null) {
@@ -242,4 +241,22 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 		return null;
 	}
 
+
+	private class EndpointReplyHandler implements ReplyHandler {
+
+		public void handle(Message<?> replyMessage, MessageHeader originalMessageHeader) {
+			if (replyMessage == null) {
+				return;
+			}
+			String replyChannelName = originalMessageHeader.getReplyChannelName();
+			MessageChannel replyChannel = resolveReplyChannel(replyChannelName);
+			if (replyChannel == null) {
+				throw new MessageHandlingException("Unable to determine reply channel for message. "
+						+ "Provide a 'replyChannelName' in the message header or a 'defaultOutputChannelName' "
+						+ "on the message endpoint.");
+			}
+			replyChannel.send(replyMessage);
+		}
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/handler/ConcurrentHandler.java

@@ -21,8 +21,6 @@ import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.Lifecycle;
-import org.springframework.integration.dispatcher.MessageHandlerNotRunningException;
-import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.util.ErrorHandler;
@@ -51,6 +49,8 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin
 
 	private ErrorHandler errorHandler;
 
+	private ReplyHandler replyHandler;
+
 	private volatile boolean running;
 
 	private Object lifecycleMonitor = new Object();
@@ -83,6 +83,10 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin
 		this.errorHandler = errorHandler;
 	}
 
+	public void setReplyHandler(ReplyHandler replyHandler) {
+		this.replyHandler = replyHandler;
+	}
+
 	public void afterPropertiesSet() {
 		refreshExecutor();
 	}
@@ -173,7 +177,10 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin
 
 		public void run() {
 			try {
-				handler.handle(this.message);
+				Message<?> reply = handler.handle(this.message);
+				if (replyHandler != null) {
+					replyHandler.handle(reply, this.message.getHeader());
+				}
 			}
 			catch (Throwable t) {
 				if (errorHandler != null) {

spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerNotRunningException.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.dispatcher;
+package org.springframework.integration.handler;
 
 import org.springframework.integration.MessageHandlingException;
 

spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerRejectedExecutionException.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.dispatcher;
+package org.springframework.integration.handler;
 
 import org.springframework.integration.MessageHandlingException;
 

spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyHandler.java

@@ -0,0 +1,31 @@
+/*
+ * 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.handler;
+
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageHeader;
+
+/**
+ * Strategy interface for handling reply messages.
+ * 
+ * @author Mark Fisher
+ */
+public interface ReplyHandler {
+
+	void handle(Message<?> replyMessage, MessageHeader originalMessageHeader);
+
+}

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

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

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

@@ -120,6 +120,7 @@ public class ChannelParserTests {
 		assertEquals(DispatcherPolicy.DEFAULT_RECEIVE_TIMEOUT, dispatcherPolicy.getReceiveTimeout());
 		assertEquals(DispatcherPolicy.DEFAULT_REJECTION_LIMIT, dispatcherPolicy.getRejectionLimit());
 		assertEquals(DispatcherPolicy.DEFAULT_RETRY_INTERVAL, dispatcherPolicy.getRetryInterval());
+		assertTrue(dispatcherPolicy.getShouldFailOnRejectionLimit());
 	}
 
 	@Test
@@ -133,6 +134,7 @@ public class ChannelParserTests {
 		assertEquals(77, dispatcherPolicy.getReceiveTimeout());
 		assertEquals(777, dispatcherPolicy.getRejectionLimit());
 		assertEquals(7777, dispatcherPolicy.getRetryInterval());
+		assertFalse(dispatcherPolicy.getShouldFailOnRejectionLimit());
 	}
 
 

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

@@ -16,7 +16,11 @@
 	<channel id="publishSubscribeChannel" publish-subscribe="true"/>
 
 	<channel id="channelWithDispatcherPolicy" publish-subscribe="true">
-		<dispatcher-policy max-messages-per-task="7" receive-timeout="77" rejection-limit="777" retry-interval="7777"/>
+		<dispatcher-policy max-messages-per-task="7"
+		                   receive-timeout="77"
+		                   rejection-limit="777"
+		                   retry-interval="7777"
+		                   should-fail-on-rejection-limit="false"/>
 	</channel>
 
 </beans:beans>

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

@@ -0,0 +1,100 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Collection;
+import java.util.Iterator;
+
+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;
+
+/**
+ * @author Mark Fisher
+ */
+public class ChannelPollingMessageRetrieverTests {
+
+	@Test
+	public void testSingleMessagePerRetrieval() {
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setReceiveTimeout(0);
+		MessageChannel channel = new SimpleChannel(dispatcherPolicy);
+		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
+		Collection<Message<?>> results = retriever.retrieveMessages();
+		assertTrue(results.isEmpty());
+		channel.send(new StringMessage("test1"));
+		channel.send(new StringMessage("test2"));
+		results = retriever.retrieveMessages();
+		assertEquals(1, results.size());
+		assertEquals("test1", results.iterator().next().getPayload());
+		results = retriever.retrieveMessages();
+		assertEquals(1, results.size());
+		assertEquals("test2", results.iterator().next().getPayload());
+	}
+
+	@Test
+	public void testMultipleMessagesPerRetrieval() {
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setReceiveTimeout(0);
+		dispatcherPolicy.setMaxMessagesPerTask(2);
+		MessageChannel channel = new SimpleChannel(dispatcherPolicy);
+		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
+		Collection<Message<?>> results = retriever.retrieveMessages();
+		assertTrue(results.isEmpty());
+		channel.send(new StringMessage("test1"));
+		channel.send(new StringMessage("test2"));
+		channel.send(new StringMessage("test3"));
+		results = retriever.retrieveMessages();
+		assertEquals(2, results.size());
+		Iterator<Message<?>> iter = results.iterator();
+		assertEquals("test1", iter.next().getPayload());
+		assertEquals("test2", iter.next().getPayload());
+		results = retriever.retrieveMessages();
+		assertEquals(1, results.size());
+		assertEquals("test3", results.iterator().next().getPayload());
+	}
+
+	@Test
+	public void testMaxMessagesConfiguredDynamically() {
+		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+		dispatcherPolicy.setReceiveTimeout(0);
+		dispatcherPolicy.setMaxMessagesPerTask(1);
+		MessageChannel channel = new SimpleChannel(dispatcherPolicy);
+		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
+		Collection<Message<?>> results = retriever.retrieveMessages();
+		assertTrue(results.isEmpty());
+		channel.send(new StringMessage("test1"));
+		channel.send(new StringMessage("test2"));
+		channel.send(new StringMessage("test3"));
+		results = retriever.retrieveMessages();
+		assertEquals(1, results.size());
+		assertEquals("test1", results.iterator().next().getPayload());
+		channel.getDispatcherPolicy().setMaxMessagesPerTask(5);
+		results = retriever.retrieveMessages();
+		assertEquals(2, results.size());
+		Iterator<Message<?>> iter = results.iterator();
+		assertEquals("test2", iter.next().getPayload());
+		assertEquals("test3", iter.next().getPayload());
+	}
+
+}

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

@@ -29,12 +29,12 @@ import org.junit.Test;
 import org.springframework.integration.MessageDeliveryException;
 import org.springframework.integration.channel.SimpleChannel;
 import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
-import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.endpoint.DefaultMessageEndpoint;
 import org.springframework.integration.handler.ConcurrentHandler;
 import org.springframework.integration.handler.InterceptingMessageHandler;
 import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
 import org.springframework.integration.handler.TestHandlers;
 import org.springframework.integration.message.ErrorMessage;
 import org.springframework.integration.message.Message;

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

@@ -0,0 +1,79 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+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 {
+
+	@Test
+	public void testSimpleDispatch() throws InterruptedException {
+		MessageChannel channel = new SimpleChannel();
+		DispatcherTask task = new DispatcherTask(channel);
+		final CountDownLatch latch = new CountDownLatch(1);
+		task.addHandler(TestHandlers.countDownHandler(latch));
+		task.dispatchMessage(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);
+		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"));
+		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);
+		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"));
+		latch.await(500, TimeUnit.MILLISECONDS);
+		assertEquals(0, latch.getCount());
+		assertEquals(1, counter1.get());
+		assertEquals(1, counter2.get());
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java

@@ -18,6 +18,12 @@ package org.springframework.integration.endpoint;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 
@@ -25,9 +31,15 @@ import org.springframework.integration.channel.ChannelRegistry;
 import org.springframework.integration.channel.DefaultChannelRegistry;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
+import org.springframework.integration.handler.ConcurrentHandler;
 import org.springframework.integration.handler.MessageHandler;
+import org.springframework.integration.handler.MessageHandlerNotRunningException;
+import org.springframework.integration.handler.TestHandlers;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.StringMessage;
+import org.springframework.integration.message.selector.MessageSelector;
+import org.springframework.integration.message.selector.MessageSelectorRejectedException;
+import org.springframework.integration.util.ErrorHandler;
 
 /**
  * @author Mark Fisher
@@ -79,4 +91,316 @@ public class DefaultMessageEndpointTests {
 		assertEquals("hello test", reply.getPayload());
 	}
 
+	@Test
+	public void testCustomErrorHandler() throws InterruptedException {
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1));
+		final CountDownLatch latch = new CountDownLatch(2);
+		endpoint.setErrorHandler(new ErrorHandler() {
+			public void handle(Throwable t) {
+				latch.countDown();
+			}
+		});
+		endpoint.setHandler(TestHandlers.rejectingCountDownHandler(latch));
+		endpoint.start();
+		endpoint.handle(new StringMessage("test"));
+		latch.await(500, TimeUnit.MILLISECONDS);
+		assertEquals("both handler and errorHandler should have been invoked", 0, latch.getCount());
+	}
+
+	@Test
+	public void testConcurrentHandlerWithDefaultReplyChannel() throws InterruptedException {
+		MessageChannel replyChannel = new SimpleChannel();
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		channelRegistry.registerChannel("replyChannel", replyChannel);
+		final CountDownLatch latch = new CountDownLatch(1);
+		MessageHandler handler = new MessageHandler() {
+			public Message<String> handle(Message<?> message) {
+				latch.countDown();
+				return new StringMessage("123", "hello " + message.getPayload());
+			}
+		};
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		endpoint.setChannelRegistry(channelRegistry);
+		endpoint.setHandler(new ConcurrentHandler(handler));
+		endpoint.setDefaultOutputChannelName("replyChannel");
+		endpoint.start();
+		endpoint.handle(new StringMessage(1, "test"));
+		endpoint.stop();
+		latch.await(200, TimeUnit.MILLISECONDS);
+		assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
+		Message<?> reply = replyChannel.receive(100);
+		assertNotNull(reply);
+		assertEquals("hello test", reply.getPayload());
+	}
+
+	@Test
+	public void testHandlerReturnsNull() throws InterruptedException {
+		MessageChannel replyChannel = new SimpleChannel();
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		channelRegistry.registerChannel("replyChannel", replyChannel);
+		final CountDownLatch latch = new CountDownLatch(1);
+		MessageHandler handler = new MessageHandler() {
+			public Message<String> handle(Message<?> message) {
+				latch.countDown();
+				return null;
+			}
+		};
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		endpoint.setChannelRegistry(channelRegistry);
+		endpoint.setHandler(handler);
+		endpoint.setDefaultOutputChannelName("replyChannel");
+		endpoint.start();
+		endpoint.handle(new StringMessage(1, "test"));
+		endpoint.stop();
+		latch.await(200, TimeUnit.MILLISECONDS);
+		assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
+		Message<?> reply = replyChannel.receive(0);
+		assertNull(reply);
+	}
+
+	@Test
+	public void testConcurrentHandlerReturnsNull() throws InterruptedException {
+		MessageChannel replyChannel = new SimpleChannel();
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		channelRegistry.registerChannel("replyChannel", replyChannel);
+		final CountDownLatch latch = new CountDownLatch(1);
+		MessageHandler handler = new MessageHandler() {
+			public Message<String> handle(Message<?> message) {
+				latch.countDown();
+				return null;
+			}
+		};
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		endpoint.setChannelRegistry(channelRegistry);
+		endpoint.setHandler(new ConcurrentHandler(handler));
+		endpoint.setDefaultOutputChannelName("replyChannel");
+		endpoint.start();
+		endpoint.handle(new StringMessage(1, "test"));
+		endpoint.stop();
+		latch.await(200, TimeUnit.MILLISECONDS);
+		assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
+		Message<?> reply = replyChannel.receive(0);
+		assertNull(reply);
+	}
+
+	@Test
+	public void testConcurrentHandlerWithExplicitReplyChannel() throws InterruptedException {
+		MessageChannel replyChannel = new SimpleChannel();
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		channelRegistry.registerChannel("replyChannel", replyChannel);
+		final CountDownLatch latch = new CountDownLatch(1);
+		MessageHandler handler = new MessageHandler() {
+			public Message<String> handle(Message<?> message) {
+				latch.countDown();
+				return new StringMessage("123", "hello " + message.getPayload());
+			}
+		};
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		endpoint.setChannelRegistry(channelRegistry);
+		endpoint.setHandler(new ConcurrentHandler(handler));
+		endpoint.start();
+		StringMessage message = new StringMessage(1, "test");
+		message.getHeader().setReplyChannelName("replyChannel");
+		endpoint.handle(message);
+		endpoint.stop();
+		latch.await(200, TimeUnit.MILLISECONDS);
+		assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
+		Message<?> reply = replyChannel.receive(100);
+		assertNotNull(reply);
+		assertEquals("hello test", reply.getPayload());
+	}
+
+	@Test
+	public void testGeneratedConcurrentHandlerWithDefaultReplyChannel() throws InterruptedException {
+		MessageChannel replyChannel = new SimpleChannel();
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		channelRegistry.registerChannel("replyChannel", replyChannel);
+		final CountDownLatch latch = new CountDownLatch(1);
+		MessageHandler handler = new MessageHandler() {
+			public Message<String> handle(Message<?> message) {
+				latch.countDown();
+				return new StringMessage("123", "hello " + message.getPayload());
+			}
+		};
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		endpoint.setChannelRegistry(channelRegistry);
+		endpoint.setHandler(handler);
+		endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(3, 14));
+		endpoint.setDefaultOutputChannelName("replyChannel");
+		endpoint.start();
+		endpoint.handle(new StringMessage(1, "test"));
+		endpoint.stop();
+		latch.await(200, TimeUnit.MILLISECONDS);
+		assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
+		Message<?> reply = replyChannel.receive(100);
+		assertNotNull(reply);
+		assertEquals("hello test", reply.getPayload());
+	}
+
+	@Test
+	public void testGeneratedConcurrentHandlerWithExplicitReplyChannel() throws InterruptedException {
+		MessageChannel replyChannel = new SimpleChannel();
+		ChannelRegistry channelRegistry = new DefaultChannelRegistry();
+		channelRegistry.registerChannel("replyChannel", replyChannel);
+		final CountDownLatch latch = new CountDownLatch(1);
+		MessageHandler handler = new MessageHandler() {
+			public Message<String> handle(Message<?> message) {
+				latch.countDown();
+				return new StringMessage("123", "hello " + message.getPayload());
+			}
+		};
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		endpoint.setChannelRegistry(channelRegistry);
+		endpoint.setHandler(handler);
+		endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(3, 14));
+		endpoint.start();
+		StringMessage message = new StringMessage(1, "test");
+		message.getHeader().setReplyChannelName("replyChannel");
+		endpoint.handle(message);
+		endpoint.stop();
+		latch.await(200, TimeUnit.MILLISECONDS);
+		assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
+		Message<?> reply = replyChannel.receive(100);
+		assertNotNull(reply);
+		assertEquals("hello test", reply.getPayload());
+	}
+
+	@Test(expected=MessageHandlerNotRunningException.class)
+	public void testEndpointDoesNotHandleMessagesWhenNotYetStarted() {
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		endpoint.handle(new StringMessage("test"));
+	}
+
+	@Test
+	public void testEndpointDoesNotHandleMessagesAfterBeingStopped() {
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		AtomicInteger counter = new AtomicInteger();
+		boolean exceptionThrown = false;
+		try {
+			endpoint.start();
+			endpoint.setHandler(TestHandlers.countingHandler(counter));
+			endpoint.handle(new StringMessage("test1"));
+			endpoint.stop();
+			endpoint.handle(new StringMessage("test2"));
+		}
+		catch (MessageHandlerNotRunningException e) {
+			exceptionThrown = true;
+		}
+		assertEquals("handler should have been invoked exactly once", 1, counter.get());
+		assertTrue(exceptionThrown);
+	}
+
+	@Test(expected=MessageSelectorRejectedException.class)
+	public void testEndpointWithSelectorRejecting() {
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		endpoint.addMessageSelector(new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				return false;
+			}
+		});
+		endpoint.start();
+		endpoint.handle(new StringMessage("test"));
+	}
+
+	@Test
+	public void testEndpointWithSelectorAccepting() throws InterruptedException {
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		endpoint.addMessageSelector(new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				return true;
+			}
+		});
+		CountDownLatch latch = new CountDownLatch(1);
+		endpoint.setHandler(TestHandlers.countDownHandler(latch));
+		endpoint.start();
+		endpoint.handle(new StringMessage("test"));
+		latch.await(100, TimeUnit.MILLISECONDS);
+		assertEquals("handler should have been invoked", 0, latch.getCount());
+		endpoint.stop();
+	}
+
+	@Test
+	public void testEndpointWithMultipleSelectorsAndFirstRejects() {
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		boolean exceptionThrown = false;
+		final AtomicInteger counter = new AtomicInteger();
+		endpoint.addMessageSelector(new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				counter.incrementAndGet();
+				return false;
+			}
+		});
+		endpoint.addMessageSelector(new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				counter.incrementAndGet();
+				return true;
+			}
+		});
+		endpoint.setHandler(TestHandlers.countingHandler(counter));
+		endpoint.start();
+		try {
+			endpoint.handle(new StringMessage("test"));
+		}
+		catch (MessageSelectorRejectedException e) {
+			exceptionThrown = true;
+		}
+		assertEquals("only the first selector should have been invoked", 1, counter.get());
+		assertTrue(exceptionThrown);
+		endpoint.stop();
+	}
+
+	@Test
+	public void testEndpointWithMultipleSelectorsAndFirstAccepts() {
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		boolean exceptionThrown = false;
+		final AtomicInteger counter = new AtomicInteger();
+		endpoint.addMessageSelector(new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				counter.incrementAndGet();
+				return true;
+			}
+		});
+		endpoint.addMessageSelector(new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				counter.incrementAndGet();
+				return false;
+			}
+		});
+		endpoint.setHandler(TestHandlers.countingHandler(counter));
+		endpoint.start();
+		try {
+			endpoint.handle(new StringMessage("test"));
+		}
+		catch (MessageSelectorRejectedException e) {
+			exceptionThrown = true;
+		}
+		assertEquals("both selectors should have been invoked but not the handler", 2, counter.get());
+		assertTrue(exceptionThrown);
+		endpoint.stop();
+	}
+
+	@Test
+	public void testEndpointWithMultipleSelectorsAndBothAccept() {
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		final AtomicInteger counter = new AtomicInteger();
+		endpoint.addMessageSelector(new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				counter.incrementAndGet();
+				return true;
+			}
+		});
+		endpoint.addMessageSelector(new MessageSelector() {
+			public boolean accept(Message<?> message) {
+				counter.incrementAndGet();
+				return true;
+			}
+		});
+		endpoint.setHandler(TestHandlers.countingHandler(counter));
+		endpoint.start();
+		endpoint.handle(new StringMessage("test"));
+		assertEquals("both selectors and handler should have been invoked", 3, counter.get());
+		endpoint.stop();
+	}
+
 }

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

@@ -19,7 +19,6 @@ 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;
 
 /**