Refactored so that endpoint is reponsible for managing ConcurrentHandlers rather than dispatchers having that responsibility. Also providing errorHandler for endpoints. That errorHandler is also used for asynchronous handler task execution.

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

@@ -42,7 +42,6 @@ import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.endpoint.DefaultMessageEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.handler.MessageHandler;
-import org.springframework.integration.handler.PooledMessageHandler;
 import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
 import org.springframework.integration.scheduling.MessagingTaskScheduler;
 import org.springframework.integration.scheduling.MessagingTaskSchedulerAware;
@@ -187,6 +186,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		if (!this.initialized) {
 			this.initialize();
 		}
+		channel.setName(name);
 		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.setMessagingTaskScheduler(this.taskScheduler);
 		if (dispatcherPolicy != null) {
@@ -213,11 +213,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		Assert.notNull(name, "'name' must not be null");
 		Assert.notNull(handler, "'handler' must not be null");
 		Assert.notNull(subscription, "'subscription' must not be null");
-		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handler);
 		endpoint.setName(name);
-		endpoint.setHandler(handler);
 		endpoint.setSubscription(subscription);
 		endpoint.setConcurrencyPolicy(concurrencyPolicy);
+		endpoint.afterPropertiesSet();
 		this.registerEndpoint(name, endpoint);
 	}
 
@@ -260,7 +260,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 			}
 		}
 		if (endpoint instanceof DefaultMessageEndpoint) {
-			String outputChannelName = ((DefaultMessageEndpoint) endpoint).getDefaultOutputChannelName();
+			DefaultMessageEndpoint dme = (DefaultMessageEndpoint) endpoint;
+			String outputChannelName = dme.getDefaultOutputChannelName();
 			if (outputChannelName != null && this.lookupChannel(outputChannelName) == null) {
 				if (!this.autoCreateChannels) {
 					throw new MessagingConfigurationException("Unknown channel '" + outputChannelName +
@@ -269,8 +270,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 				}
 				this.registerChannel(outputChannelName, new SimpleChannel());
 			}
+			if (!dme.hasErrorHandler() && this.getErrorChannel() != null) {
+				dme.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel()));
+			}
 		}
-		this.registerWithDispatcher(channel, endpoint, subscription.getSchedule(), endpoint.getConcurrencyPolicy());
+		this.registerWithDispatcher(channel, endpoint, subscription.getSchedule());
 		if (logger.isInfoEnabled()) {
 			logger.info("activated subscription to channel '" + channel.getName() + 
 					"' for endpoint '" + endpoint.getName() + "'");
@@ -295,16 +299,13 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 	}
 
-	private void registerWithDispatcher(MessageChannel channel, MessageHandler handler, Schedule schedule, ConcurrencyPolicy concurrencyPolicy) {
+	private void registerWithDispatcher(MessageChannel channel, MessageHandler handler, Schedule schedule) {
 		MessageDispatcher 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");
 			}
 		}
-		if (concurrencyPolicy != null) {
-			handler = new PooledMessageHandler(handler, concurrencyPolicy.getCoreSize(), concurrencyPolicy.getMaxSize());
-		}
 		dispatcher.addHandler(handler, schedule);
 		if (this.isRunning() && !dispatcher.isRunning()) {
 			dispatcher.start();
@@ -349,23 +350,27 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 	}
 
 	public void stop() {
+		if (!this.isRunning()) {
+			return;
+		}
 		synchronized (this.lifecycleMonitor) {
-			if (this.isRunning()) {
-				this.running = false;
-				this.taskScheduler.stop();
-				for (Lifecycle adapter : this.lifecycleSourceAdapters) {
-					adapter.stop();
-					if (logger.isInfoEnabled()) {
-						logger.info("stopped source adapter '" + adapter + "'");
-					}
-				}
-				for (MessageDispatcher dispatcher : this.dispatchers.values()) {
-					dispatcher.stop();
-					if (logger.isInfoEnabled()) {
-						logger.info("stopped dispatcher '" + dispatcher + "'");
-					}
+			this.running = false;
+			this.taskScheduler.stop();
+			for (Lifecycle adapter : this.lifecycleSourceAdapters) {
+				adapter.stop();
+				if (logger.isInfoEnabled()) {
+					logger.info("stopped source adapter '" + adapter + "'");
 				}
 			}
+			for (MessageDispatcher dispatcher : this.dispatchers.values()) {
+				dispatcher.stop();
+				if (logger.isInfoEnabled()) {
+					logger.info("stopped dispatcher '" + dispatcher + "'");
+				}
+			}
+		}
+		if (logger.isInfoEnabled()) {
+			logger.info("message bus stopped");
 		}
 	}
 

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

@@ -30,6 +30,11 @@ public interface MessageChannel {
 	 */
 	String getName();
 
+	/**
+	 * Set the name of this channel.
+	 */
+	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

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

@@ -156,51 +156,53 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas
 		if (!this.scheduler.isRunning()) {
 			this.scheduler.start();
 		}
+		if (this.isRunning()) {
+			return;
+		}
 		synchronized (this.lifecycleMonitor) {
-			if (!this.isRunning()) {
-				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);
-					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();
-						}
-						task.addHandler(handler);
-					}
-					ScheduledFuture<?> future = this.scheduler.schedule(task);
-					if (future != null) {
-						futures.add(future);
+			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);
+				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();
 					}
+					task.addHandler(handler);
+				}
+				ScheduledFuture<?> future = this.scheduler.schedule(task);
+				if (future != null) {
+					futures.add(future);
 				}
-				this.running = true;
 			}
+			this.running = true;
 		}
 	}
 
 	public void stop() {
+		if (!this.isRunning()) {
+			return;
+		}
 		synchronized (this.lifecycleMonitor) {
-			if (this.isRunning()) {
-				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 (ScheduledFuture<?> future : this.futures) {
+				future.cancel(true);
+				for (List<MessageHandler> handlerList : scheduledHandlers.values()) {
+					for (MessageHandler handler : handlerList) {
+						if (handler instanceof Lifecycle) {
+							((Lifecycle) handler).stop();
 						}
 					}
 				}
-				this.running = false;
 			}
+			this.running = false;
 		}
 	}
 

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

@@ -26,6 +26,10 @@ import org.springframework.integration.MessageHandlingException;
  */
 public class MessageSelectorRejectedException extends MessageHandlingException {
 
+	public MessageSelectorRejectedException() {
+		super();
+	}
+
 	public MessageSelectorRejectedException(String message) {
 		super(message);
 	}

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

@@ -30,6 +30,15 @@ public class ConcurrencyPolicy implements EndpointPolicy {
 	private int maxSize;
 
 
+	public ConcurrencyPolicy() {
+	}
+
+	public ConcurrencyPolicy(int coreSize, int maxSize) {
+		this.setCoreSize(coreSize);
+		this.setMaxSize(maxSize);
+	}
+
+
 	public int getCoreSize() {
 		return this.coreSize;
 	}

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

@@ -16,35 +16,67 @@
 
 package org.springframework.integration.endpoint;
 
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.Lifecycle;
 import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.MessagingConfigurationException;
 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.MessageSelectorRejectedException;
+import org.springframework.integration.handler.ConcurrentHandler;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.selector.MessageSelector;
+import org.springframework.integration.util.ErrorHandler;
+import org.springframework.util.Assert;
 
 /**
  * Default implementation of the {@link MessageEndpoint} interface.
  * 
  * @author Mark Fisher
  */
-public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryAware, BeanNameAware {
+public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryAware, InitializingBean, BeanNameAware {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
 
 	private String name;
 
 	private MessageHandler handler;
 
+	private List<MessageSelector> selectors = new CopyOnWriteArrayList<MessageSelector>();
+
 	private Subscription subscription;
 
 	private ConcurrencyPolicy concurrencyPolicy;
 
+	private ErrorHandler errorHandler;
+
 	private String defaultOutputChannelName;
 
 	private ChannelRegistry channelRegistry;
 
+	private volatile boolean initialized;
+
+	private volatile boolean running;
+
+
+	public DefaultMessageEndpoint() {
+	}
+
+	public DefaultMessageEndpoint(MessageHandler handler) {
+		this.handler = handler;
+	}
+
 
 	public String getName() {
 		return this.name;
@@ -58,28 +90,22 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 		this.setName(beanName);
 	}
 
-	public String getDefaultOutputChannelName() {
-		return this.defaultOutputChannelName;
-	}
-
-	/**
-	 * Set the name of the channel to which this endpoint can send reply messages by default.
-	 */
-	public void setDefaultOutputChannelName(String defaultOutputChannelName) {
-		this.defaultOutputChannelName = defaultOutputChannelName;
-	}
-
 	public MessageHandler getHandler() {
 		return this.handler;
 	}
 
 	/**
-	 * Set a handler to be invoked for each consumed message.
+	 * Set the handler to be invoked for each consumed message.
 	 */
 	public void setHandler(MessageHandler handler) {
 		this.handler = handler;
 	}
 
+	public void addMessageSelector(MessageSelector messageSelector) {
+		Assert.notNull(messageSelector, "'messageSelector' must not be null");
+		this.selectors.add(messageSelector);
+	}
+
 	public Subscription getSubscription() {
 		return this.subscription;
 	}
@@ -96,6 +122,25 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 		this.concurrencyPolicy = concurrencyPolicy;
 	}
 
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
+	public boolean hasErrorHandler() {
+		return (this.errorHandler != null);
+	}
+
+	public String getDefaultOutputChannelName() {
+		return this.defaultOutputChannelName;
+	}
+
+	/**
+	 * Set the name of the channel to which this endpoint can send reply messages by default.
+	 */
+	public void setDefaultOutputChannelName(String defaultOutputChannelName) {
+		this.defaultOutputChannelName = defaultOutputChannelName;
+	}
+
 	/**
 	 * Set the channel registry to use for looking up channels by name.
 	 */
@@ -103,7 +148,55 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 		this.channelRegistry = channelRegistry;
 	}
 
-	public Message<?> handle(Message<?> message) {
+	public void afterPropertiesSet() {
+		if (this.concurrencyPolicy != null) {
+			if (!(this.handler instanceof ConcurrentHandler)) {
+				this.handler = new ConcurrentHandler(this.handler);
+			}
+			ConcurrentHandler concurrentHandler = (ConcurrentHandler) this.handler;
+			concurrentHandler.setCorePoolSize(this.concurrencyPolicy.getCoreSize());
+			concurrentHandler.setMaxPoolSize(this.concurrencyPolicy.getMaxSize());
+			if (this.errorHandler != null) {
+				concurrentHandler.setErrorHandler(this.errorHandler);
+			}
+			concurrentHandler.afterPropertiesSet();
+		}
+		this.initialized = true;
+	}
+
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	public void start() {
+		if (this.isRunning()) {
+			return;
+		}
+		if (this.handler instanceof Lifecycle) {
+			((Lifecycle) handler).start();
+		}
+		this.running = true;
+	}
+
+	public void stop() {
+		if (!this.isRunning()) {
+			return;
+		}
+		if (this.handler instanceof Lifecycle) {
+			((Lifecycle) handler).stop();
+		}
+		this.running = false;
+	}
+
+	public final Message<?> handle(Message<?> message) {
+		if (!this.isRunning()) {
+			throw new MessageHandlerNotRunningException();
+		}
+		for (MessageSelector selector : this.selectors) {
+			if (!selector.accept(message)) {
+				throw new MessageSelectorRejectedException();
+			}
+		}
 		if (this.handler == null) {
 			if (this.defaultOutputChannelName == null) {
 				throw new MessagingConfigurationException(
@@ -113,15 +206,24 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 			replyChannel.send(message);
 			return null;
 		}
-		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.");
+		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);
 			}
-			replyChannel.send(replyMessage);
+		}
+		catch (Throwable t) {
+			if (this.errorHandler == null) {
+				throw new MessageHandlingException(
+						"error occurred in endpoint, and no 'errorHandler' available", t);
+			}
+			this.errorHandler.handle(t);
 		}
 		return null;
 	}

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.endpoint;
 
+import org.springframework.context.Lifecycle;
 import org.springframework.integration.bus.Subscription;
 import org.springframework.integration.handler.MessageHandler;
 
@@ -24,7 +25,7 @@ import org.springframework.integration.handler.MessageHandler;
  * 
  * @author Mark Fisher
  */
-public interface MessageEndpoint extends MessageHandler {
+public interface MessageEndpoint extends MessageHandler, Lifecycle {
 
 	String getName();
 

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

@@ -0,0 +1,196 @@
+/*
+ * 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 java.util.concurrent.RejectedExecutionException;
+
+import org.apache.commons.logging.Log;
+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.message.Message;
+import org.springframework.integration.util.ErrorHandler;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link MessageHandler} implementation that encapsulates a
+ * {@link ThreadPoolTaskExecutor} and delegates to a wrapped handler for
+ * concurrent, asynchronous message handling.
+ * 
+ * @author Mark Fisher
+ */
+public class ConcurrentHandler implements MessageHandler, Lifecycle, InitializingBean {
+
+	private Log logger = LogFactory.getLog(this.getClass());
+
+	private MessageHandler handler;
+
+	private ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+
+	private int corePoolSize = 1;
+
+	private int maxPoolSize = 5;
+
+	private int queueCapacity = 0;
+
+	private int keepAliveSeconds = 60;
+
+	private ErrorHandler errorHandler;
+
+	private volatile boolean running;
+
+	private Object lifecycleMonitor = new Object();
+
+
+	public ConcurrentHandler(MessageHandler handler) {
+		Assert.notNull(handler, "'handler' must not be null");
+		this.handler = handler;
+	}
+
+	public ConcurrentHandler(MessageHandler handler, int corePoolSize, int maxPoolSize) {
+		Assert.notNull(handler, "'handler' must not be null");
+		Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1");
+		Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1");
+		Assert.isTrue(maxPoolSize >= corePoolSize, "'corePoolSize' cannot exceed 'maxPoolSize'");
+		this.handler = handler;
+		this.corePoolSize = corePoolSize;
+		this.maxPoolSize = maxPoolSize;
+	}
+
+
+	public void setExecutor(ThreadPoolTaskExecutor executor) {
+		Assert.notNull(executor, "'executor' must not be null");
+		this.executor = executor;
+	}
+
+	public void setCorePoolSize(int corePoolSize) {
+		Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1");
+		this.corePoolSize = corePoolSize;
+		if (this.executor != null) {
+			this.executor.setCorePoolSize(corePoolSize);
+		}
+	}
+
+	public void setMaxPoolSize(int maxPoolSize) {
+		Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1");
+		this.maxPoolSize = maxPoolSize;
+		if (this.executor != null) {
+			this.executor.setMaxPoolSize(maxPoolSize);
+		}
+	}
+
+	public void setQueueCapacity(int queueCapacity) {
+		this.queueCapacity = queueCapacity;
+		if (this.executor != null) {
+			this.executor.setQueueCapacity(queueCapacity);
+		}
+	}
+
+	public void setKeepAliveSeconds(int keepAliveSeconds) {
+		this.keepAliveSeconds = keepAliveSeconds;
+		if (this.executor != null) {
+			this.executor.setKeepAliveSeconds(keepAliveSeconds);
+		}
+	}
+
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
+	public void afterPropertiesSet() {
+		initializeExecutor();
+	}
+
+	private void initializeExecutor() {
+		if (this.executor == null) {
+			this.executor = new ThreadPoolTaskExecutor();
+		}
+		this.executor.setCorePoolSize(this.corePoolSize);
+		this.executor.setMaxPoolSize(this.maxPoolSize);
+		this.executor.setQueueCapacity(this.queueCapacity);
+		this.executor.setKeepAliveSeconds(this.keepAliveSeconds);
+		CustomizableThreadFactory threadFactory = new CustomizableThreadFactory();
+		threadFactory.setThreadNamePrefix("handler-");
+		this.executor.setThreadFactory(threadFactory);
+		this.executor.afterPropertiesSet();
+	}
+
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	public void start() {
+		synchronized (this.lifecycleMonitor) {
+			if (!this.running) {
+				this.afterPropertiesSet();
+			}
+			this.running = true;
+		}
+	}
+
+	public void stop() {
+		synchronized (this.lifecycleMonitor) {
+			if (this.isRunning()) {
+				this.executor.shutdown();
+			}
+			this.running = false;
+		}
+	}
+
+	public Message<?> handle(Message<?> message) {
+		if (!this.isRunning()) {
+			throw new MessageHandlerNotRunningException();
+		}
+		try {
+			this.executor.execute(new HandlerTask(message));
+			return null;
+		}
+		catch (RejectedExecutionException e) {
+			throw new MessageHandlerRejectedExecutionException(e);
+		}
+	}
+
+
+	private class HandlerTask implements Runnable {
+
+		private Message<?> message;
+
+		HandlerTask(Message<?> message) {
+			this.message = message;
+		}
+
+		public void run() {
+			try {
+				handler.handle(this.message);
+			}
+			catch (Throwable t) {
+				if (errorHandler != null) {
+					errorHandler.handle(t);
+				}
+				else if (logger.isWarnEnabled()) {
+					logger.warn("error occurred in handler execution", t);
+				}
+			}
+		}
+	}
+
+}

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

@@ -34,7 +34,7 @@ public abstract class InterceptingMessageHandler implements MessageHandler {
 		this.target = target;
 	}
 
-	public Message<?> handle(Message<?> message) {
+	public final Message<?> handle(Message<?> message) {
 		return handle(message, this.target);
 	}
 

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

@@ -1,225 +0,0 @@
-/*
- * 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 java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.springframework.context.Lifecycle;
-import org.springframework.integration.dispatcher.MessageHandlerNotRunningException;
-import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException;
-import org.springframework.integration.dispatcher.MessageSelectorRejectedException;
-import org.springframework.integration.message.Message;
-import org.springframework.integration.message.selector.MessageSelector;
-import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
-import org.springframework.util.Assert;
-
-/**
- * Encapsulates a {@link ThreadPoolExecutor} with configurable error thresholds.
- * 
- * @author Mark Fisher
- */
-public class PooledMessageHandler implements MessageHandler, Lifecycle {
-
-	private Log logger = LogFactory.getLog(this.getClass());
-
-	private MessageHandler handler;
-
-	private List<MessageSelector> selectors = new CopyOnWriteArrayList<MessageSelector>();
-
-	private ThreadPoolExecutor threadPoolExecutor;
-
-	private int corePoolSize;
-
-	private int maxPoolSize;
-
-	private volatile boolean running;
-
-	private Object lifecycleMonitor = new Object();
-
-	private int successiveErrorCount;
-
-	private int successiveErrorThreshold = -1;
-
-	private int totalErrorCount;
-
-	private int totalErrorThreshold = -1;
-
-
-	public PooledMessageHandler(MessageHandler handler, int corePoolSize, int maxPoolSize) {
-		Assert.notNull(handler, "'handler' must not be null");
-		Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1");
-		Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1");
-		Assert.isTrue(maxPoolSize >= corePoolSize, "'corePoolSize' cannot exceed 'maxPoolSize'");
-		this.handler = handler;
-		this.corePoolSize = corePoolSize;
-		this.maxPoolSize = maxPoolSize;
-	}
-
-	public void setCorePoolSize(int corePoolSize) {
-		this.corePoolSize = corePoolSize;
-	}
-
-	public void setMaxPoolSize(int maxPoolSize) {
-		this.maxPoolSize = maxPoolSize;
-	}
-
-	public void addMessageSelector(MessageSelector messageSelector) {
-		Assert.notNull(messageSelector, "'messageSelector' must not be null");
-		this.selectors.add(messageSelector);
-	}
-
-	public boolean isRunning() {
-		return this.running;
-	}
-
-	public void start() {
-		synchronized (this.lifecycleMonitor) {
-			if (!this.running) {
-				this.threadPoolExecutor = new MessageHandlerThreadPoolExecutor(this.corePoolSize, this.maxPoolSize);
-			}
-			this.running = true;
-		}
-	}
-
-	public void stop() {
-		synchronized (this.lifecycleMonitor) {
-			if (this.isRunning()) {
-				this.threadPoolExecutor.shutdown();
-				this.threadPoolExecutor = null;
-			}
-			this.running = false;
-		}
-	}
-
-	public Message<?> handle(Message<?> message) {
-		if (!this.isRunning()) {
-			throw new MessageHandlerNotRunningException();
-		}
-		for (MessageSelector selector : this.selectors) {
-			if (!selector.accept(message)) {
-				throw new MessageSelectorRejectedException("selector rejected message");
-			}
-		}
-		try {
-			this.threadPoolExecutor.execute(new HandlerTask(this.handler, message));
-			return null;
-		}
-		catch (RejectedExecutionException e) {
-			throw new MessageHandlerRejectedExecutionException(
-					"handler executor rejected message", e);
-		}
-	}
-
-	/**
-	 * Set the maximum number of errors allowed in <em>successive</em>
-	 * executions. If this threshold is ever exceeded, the executor
-	 * will shutdown.
-	 */
-	public void setSuccessiveErrorThreshold(int successiveErrorThreshold) {
-		this.successiveErrorThreshold = successiveErrorThreshold;
-	}
-
-	/**
-	 * Set the maximum number of <em>total</em> errors allowed in executions
-	 * If this threshold is ever exceeded, the executor will shutdown.
-	 */
-	public void setTotalErrorThreshold(int totalErrorThreshold) {
-		this.totalErrorThreshold = totalErrorThreshold;
-	}
-
-	public int getActiveCount() {
-		if (this.threadPoolExecutor == null) {
-			return 0;
-		}
-		return this.threadPoolExecutor.getActiveCount();
-	}
-
-	public boolean isShutdown() {
-		return this.threadPoolExecutor.isShutdown();
-	}
-
-
-	private static class HandlerTask implements Runnable {
-
-		private MessageHandler handler;
-
-		private Message<?> message;
-
-		private Throwable error;
-
-
-		HandlerTask(MessageHandler handler, Message<?> message) {
-			this.handler = handler;
-			this.message = message;
-		}
-
-		public Throwable getError() {
-			return this.error;
-		}
-
-		public void run() {
-			try {
-				this.handler.handle(this.message);
-			}
-			catch (Throwable t) {
-				this.error = t;
-			}
-		}
-	}
-
-
-	private class MessageHandlerThreadPoolExecutor extends ThreadPoolExecutor {
-
-		public MessageHandlerThreadPoolExecutor(int corePoolSize, int maximumPoolSize) {
-			super(corePoolSize, maximumPoolSize, 0, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>());
-			CustomizableThreadFactory threadFactory = new CustomizableThreadFactory();
-			threadFactory.setThreadNamePrefix("handler-pool-");
-			this.setThreadFactory(threadFactory);
-		}
-
-		@Override
-		protected void afterExecute(Runnable r, Throwable t) {
-			HandlerTask task = (HandlerTask) r;
-			if (task.getError() != null) {
-				if (logger.isWarnEnabled()) {
-					logger.warn("Exception occurred during task execution", task.getError());
-				}
-				successiveErrorCount++;
-				totalErrorCount++;
-				if ((successiveErrorThreshold >= 0 && successiveErrorCount > successiveErrorThreshold)
-						|| (totalErrorThreshold >= 0 && totalErrorCount > totalErrorThreshold)) {
-					if (logger.isInfoEnabled()) {
-						logger.info("error threshold exceeded, shutting down now");
-					}
-					this.shutdownNow();
-				}
-			}
-			else {
-				successiveErrorCount = 0;
-			}
-		}
-	}
-
-}