Replaced ConcurrentTarget and the <concurrency/> element with ConcurrencyInterceptor.

org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java

@@ -29,6 +29,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ApplicationEvent;
@@ -47,18 +48,19 @@ import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.channel.factory.ChannelFactory;
 import org.springframework.integration.channel.factory.QueueChannelFactory;
-import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.endpoint.DefaultEndpointRegistry;
 import org.springframework.integration.endpoint.EndpointRegistry;
 import org.springframework.integration.endpoint.HandlerEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.endpoint.MessagingGateway;
+import org.springframework.integration.endpoint.MessageProducingEndpoint;
 import org.springframework.integration.endpoint.SourceEndpoint;
+import org.springframework.integration.endpoint.MessageConsumingEndpoint;
 import org.springframework.integration.endpoint.TargetEndpoint;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.CommandMessage;
-import org.springframework.integration.message.PollCommand;
 import org.springframework.integration.message.MessageTarget;
+import org.springframework.integration.message.PollCommand;
 import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
 import org.springframework.integration.scheduling.MessagingTask;
 import org.springframework.integration.scheduling.MessagingTaskScheduler;
@@ -99,8 +101,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 
 	private volatile ScheduledExecutorService executor;
 
-	private volatile ConcurrencyPolicy defaultConcurrencyPolicy;
-
 	private volatile boolean configureAsyncEventMulticaster = false;
 
 	private volatile boolean autoCreateChannels = false;
@@ -144,14 +144,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 		this.executor = executor;
 	}
 
-	/**
-	 * Specify the default concurrency policy to be used for any endpoint that
-	 * is registered without an explicitly provided policy of its own.
-	 */
-	public void setDefaultConcurrencyPolicy(ConcurrencyPolicy defaultConcurrencyPolicy) {
-		this.defaultConcurrencyPolicy = defaultConcurrencyPolicy;
-	}
-
 	/**
 	 * Set whether to automatically start the bus after initialization.
 	 * <p>Default is 'true'; set this to 'false' to allow for manual startup
@@ -220,7 +212,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 			if (this.getErrorChannel() == null) {
 				this.setErrorChannel(new DefaultErrorChannel());
 			}
-			this.taskScheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel()));
 			this.initialized = true;
 			this.initializing = false;
 		}
@@ -263,32 +254,20 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 	}
 
 	public void registerHandler(String name, MessageHandler handler, Subscription subscription) {
-		this.registerHandler(name, handler, subscription, this.defaultConcurrencyPolicy);
-	}
-
-	public void registerHandler(String name, MessageHandler handler, Subscription subscription,
-			ConcurrencyPolicy concurrencyPolicy) {
 		Assert.notNull(handler, "'handler' must not be null");
 		HandlerEndpoint endpoint = new HandlerEndpoint(handler);
-		this.doRegisterEndpoint(name, endpoint, subscription, concurrencyPolicy);
+		this.doRegisterEndpoint(name, endpoint, subscription);
 	}
 
 	public void registerTarget(String name, MessageTarget target, Subscription subscription) {
-		this.registerTarget(name, target, subscription, this.defaultConcurrencyPolicy);
-	}
-
-	public void registerTarget(String name, MessageTarget target, Subscription subscription,
-			ConcurrencyPolicy concurrencyPolicy) {
 		Assert.notNull(target, "'target' must not be null");
 		TargetEndpoint endpoint = new TargetEndpoint(target);
-		this.doRegisterEndpoint(name, endpoint, subscription, concurrencyPolicy);
+		this.doRegisterEndpoint(name, endpoint, subscription);
 	}
 
-	private void doRegisterEndpoint(String name, TargetEndpoint endpoint, Subscription subscription,
-			ConcurrencyPolicy concurrencyPolicy) {
+	private void doRegisterEndpoint(String name, TargetEndpoint endpoint, Subscription subscription) {
 		endpoint.setName(name);
 		endpoint.setSubscription(subscription);
-		endpoint.setConcurrencyPolicy(concurrencyPolicy);
 		this.registerEndpoint(name, endpoint);
 	}
 
@@ -299,10 +278,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 		if (endpoint instanceof ChannelRegistryAware) {
 			((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry);
 		}
-		if (endpoint instanceof TargetEndpoint) {
-			this.registerTargetEndpoint((TargetEndpoint) endpoint);
-		}
-		else if (endpoint instanceof SourceEndpoint) {
+		if (endpoint instanceof SourceEndpoint) {
 			this.registerSourceEndpoint(name, (SourceEndpoint) endpoint);
 		}
 		this.endpointRegistry.registerEndpoint(name, endpoint);
@@ -314,12 +290,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 		}
 	}
 
-	private void registerTargetEndpoint(TargetEndpoint endpoint) {
-		if (endpoint.getConcurrencyPolicy() == null && this.defaultConcurrencyPolicy != null) {
-			endpoint.setConcurrencyPolicy(this.defaultConcurrencyPolicy);
-		}
-	}
-
 	public MessageEndpoint unregisterEndpoint(String name) {
 		MessageEndpoint endpoint = this.endpointRegistry.unregisterEndpoint(name);
 		if (endpoint == null) {
@@ -357,22 +327,41 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 	}
 
 	private void activateEndpoint(MessageEndpoint endpoint) {
-		if (endpoint instanceof TargetEndpoint) {
-			this.activateTargetEndpoint((TargetEndpoint) endpoint);
+		if (endpoint instanceof MessageProducingEndpoint) {
+			String channelName = ((MessageProducingEndpoint) endpoint).getOutputChannelName();
+			if (channelName != null && this.lookupChannel(channelName) == null) {
+				if (!this.autoCreateChannels) {
+					throw new ConfigurationException("Unknown channel '" + channelName
+							+ "' configured as output channel for endpoint '" + endpoint
+							+ "'. Consider enabling the 'autoCreateChannels' option for the message bus.");
+				}
+				this.registerChannel(channelName, new QueueChannel());
+			}
+		}
+		if (endpoint instanceof InitializingBean) {
+			try {
+				((InitializingBean) endpoint).afterPropertiesSet();
+			}
+			catch (Exception e) {
+				throw new ConfigurationException("failed to initialize endpoint", e);
+			}
+		}
+		if (endpoint instanceof MessageConsumingEndpoint && endpoint instanceof MessageTarget) {
+			this.activateSubscriber((MessageConsumingEndpoint) endpoint);
 		}
 	}
 
-	private void activateTargetEndpoint(TargetEndpoint endpoint) {
-		Subscription subscription = endpoint.getSubscription();
+	private void activateSubscriber(MessageConsumingEndpoint subscriber) {
+		Subscription subscription = subscriber.getSubscription();
 		if (subscription == null) {
-			throw new ConfigurationException("Unable to register endpoint '" + endpoint
+			throw new ConfigurationException("Unable to register endpoint '" + subscriber
 					+ "'. No subscription information is available.");
 		}
 		MessageChannel channel = subscription.getChannel();
 		if (channel == null) {
 			String channelName = subscription.getChannelName();
 			if (channelName == null) {
-				throw new ConfigurationException("endpoint '" + endpoint
+				throw new ConfigurationException("endpoint '" + subscriber
 						+ "' must provide either 'channel' or 'channelName' in its subscription metadata");
 			}
 			channel = this.lookupChannel(channelName);
@@ -388,27 +377,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 				this.registerChannel(channelName, channel);
 			}
 		}
-		if (endpoint instanceof HandlerEndpoint) {
-			HandlerEndpoint handlerEndpoint = (HandlerEndpoint) endpoint;
-			String outputChannelName = handlerEndpoint.getOutputChannelName();
-			if (outputChannelName != null && this.lookupChannel(outputChannelName) == null) {
-				if (!this.autoCreateChannels) {
-					throw new ConfigurationException("Unknown channel '" + outputChannelName
-							+ "' configured as output channel for endpoint '" + endpoint
-							+ "'. Consider enabling the 'autoCreateChannels' option for the message bus.");
-				}
-				this.registerChannel(outputChannelName, new QueueChannel());
-			}
-		}
-		if (!endpoint.hasErrorHandler() && this.getErrorChannel() != null && !this.getErrorChannel().equals(channel)) {
-			endpoint.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel()));
-		}
-		endpoint.afterPropertiesSet();
-		this.activateSubscription(channel, endpoint, subscription.getSchedule());
+		this.activateSubscription(channel, (MessageTarget) subscriber, subscription.getSchedule());
 		if (logger.isInfoEnabled()) {
-			logger
-					.info("activated subscription to channel '" + channel.getName() + "' for endpoint '" + endpoint
-							+ "'");
+			logger.info("activated subscription to channel '" + channel.getName()
+					+ "' for endpoint '" + subscriber + "' of type '" + subscriber.getClass() + "'");
 		}
 	}
 
@@ -447,7 +419,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 		}
 	}
 
-	private void activateSubscription(MessageChannel channel, TargetEndpoint targetEndpoint, Schedule schedule) {
+	private void activateSubscription(MessageChannel channel, MessageTarget target, Schedule schedule) {
 		SubscriptionManager manager = this.subscriptionManagers.get(channel);
 		if (manager == null) {
 			if (logger.isWarnEnabled()) {
@@ -456,7 +428,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 			}
 			return;
 		}
-		manager.addTarget(targetEndpoint, schedule);
+		manager.addTarget(target, schedule);
 		if (this.isRunning() && !manager.isRunning()) {
 			manager.start();
 		}
@@ -479,6 +451,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 		this.starting = true;
 		synchronized (this.lifecycleMonitor) {
 			this.activateEndpoints();
+			this.taskScheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel()));
 			this.taskScheduler.start();
 			for (SubscriptionManager manager : this.subscriptionManagers.values()) {
 				manager.start();

org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBusAwareBeanPostProcessor.java

@@ -44,6 +44,9 @@ public class MessageBusAwareBeanPostProcessor implements BeanPostProcessor {
 	}
 
 	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+		if (bean instanceof MessageBusAware) {
+			((MessageBusAware) bean).setMessageBus(messageBus);
+		}
 		return bean;
 	}
 

org.springframework.integration/src/main/java/org/springframework/integration/bus/SubscriptionManager.java

@@ -30,13 +30,10 @@ import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.dispatcher.DirectChannel;
 import org.springframework.integration.dispatcher.PollingDispatcherTask;
-import org.springframework.integration.endpoint.TargetEndpoint;
-import org.springframework.integration.message.MessagingException;
 import org.springframework.integration.message.MessageTarget;
 import org.springframework.integration.scheduling.MessagingTaskScheduler;
 import org.springframework.integration.scheduling.PollingSchedule;
 import org.springframework.integration.scheduling.Schedule;
-import org.springframework.integration.util.ErrorHandler;
 import org.springframework.util.Assert;
 
 /**
@@ -88,7 +85,7 @@ public class SubscriptionManager {
 		}
 		else if (this.channel instanceof DirectChannel) {
 			if (logger.isInfoEnabled()) {
-				logger.info("Subscribing to a SynchronousChannel. The provided schedule will be ignored.");
+				logger.info("Subscribing to a DirectChannel. The provided schedule will be ignored.");
 			}
 		}
 		else if (this.channel.getDispatcherPolicy().isPublishSubscribe()) {
@@ -107,16 +104,6 @@ public class SubscriptionManager {
 		}
 		if (this.channel instanceof DirectChannel) {
 			((DirectChannel) this.channel).subscribe(target);
-			if (target instanceof TargetEndpoint) {
-				((TargetEndpoint) target).setErrorHandler(new ErrorHandler() {
-					public void handle(Throwable t) {
-						if (t instanceof MessagingException) {
-							throw (MessagingException) t;
-						}
-						throw new MessagingException("error occurred in handler", t);
-					}
-				});
-			}
 			return;
 		}
 		PollingDispatcherTask dispatcherTask = this.dispatcherTasks.get(schedule);

org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java

@@ -32,7 +32,9 @@ import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
+import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor;
 import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
 import org.springframework.transaction.interceptor.NoRollbackRuleAttribute;
 import org.springframework.transaction.interceptor.RollbackRuleAttribute;
@@ -102,6 +104,10 @@ public abstract class IntegrationNamespaceUtils {
 					String txInterceptorBeanName = parseTransactionInterceptor(childElement, parserContext);
 					interceptors.add(new RuntimeBeanReference(txInterceptorBeanName));
 				}
+				else if ("concurrency-interceptor".equals(localName)) {
+					String concurrencyInterceptorBeanName = parseConcurrencyInterceptor(childElement, parserContext);
+					interceptors.add(new RuntimeBeanReference(concurrencyInterceptorBeanName));
+				}
 			}
 		}
 		return interceptors;
@@ -161,6 +167,24 @@ public abstract class IntegrationNamespaceUtils {
 		return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
 	}
 
+	private static String parseConcurrencyInterceptor(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ConcurrencyInterceptor.class);
+		String taskExecutorRef = element.getAttribute("task-executor");
+		if (StringUtils.hasText(taskExecutorRef)) {
+			if (element.getAttributes().getLength() != 1) {
+				parserContext.getReaderContext().error("No other attributes are permitted when "
+					+ "specifying a 'task-executor' reference on the <concurrency-interceptor/> element.",
+					parserContext.extractSource(element));
+			}
+			builder.addConstructorArgReference(taskExecutorRef);
+		}
+		else {
+			ConcurrencyPolicy policy = IntegrationNamespaceUtils.parseConcurrencyPolicy(element);
+			builder.addConstructorArgValue(policy);
+		}
+		return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
+	}
+
 	/**
 	 * Populates the property identified by propertyName on the bean definition
 	 * to the value of the attribute specified by attributeName, if that

org.springframework.integration/src/main/java/org/springframework/integration/config/MessageEndpointBeanPostProcessor.java

@@ -46,7 +46,7 @@ public class MessageEndpointBeanPostProcessor implements BeanPostProcessor {
 			if (interceptors.size() > 0) {
 				ProxyFactory proxyFactory = new ProxyFactory(endpoint);
 				for (Advice interceptor : interceptors) {
-					proxyFactory.addAdvisor(new EndpointInvokeMethodAdvisor(interceptor));
+					proxyFactory.addAdvisor(new EndpointMethodAdvisor(interceptor));
 				}
 				bean = proxyFactory.getProxy();
 			}
@@ -56,16 +56,16 @@ public class MessageEndpointBeanPostProcessor implements BeanPostProcessor {
 
 
 	@SuppressWarnings("serial")
-	private static class EndpointInvokeMethodAdvisor extends StaticMethodMatcherPointcutAdvisor {
+	private static class EndpointMethodAdvisor extends StaticMethodMatcherPointcutAdvisor {
 
-		EndpointInvokeMethodAdvisor(Advice advice) {
+		EndpointMethodAdvisor(Advice advice) {
 			super(advice);
 		}
 
 
 		@SuppressWarnings("unchecked")
 		public boolean matches(Method method, Class clazz) {
-			return method.getName().equals("invoke")
+			return (method.getName().equals("invoke") || method.getName().equals("send"))
 					&& method.getParameterTypes().length == 1
 					&& method.getParameterTypes()[0].equals(Message.class);
 		}

org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java

@@ -41,6 +41,7 @@ import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.endpoint.HandlerEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.handler.MessageHandlerChain;
 import org.springframework.integration.handler.config.DefaultMessageHandlerCreator;
@@ -111,6 +112,7 @@ public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPost
 		return handler;
 	}
 
+	@SuppressWarnings("unchecked")
 	protected MessageHandler processResults(List<MessageHandler> results) {
 		MessageHandlerChain handlerChain = new MessageHandlerChain();
 		for (MessageHandler handler : results) {
@@ -146,7 +148,7 @@ public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPost
 					concurrencyAnnotation.coreSize(), concurrencyAnnotation.maxSize());
 			concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds());
 			concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity());
-			endpoint.setConcurrencyPolicy(concurrencyPolicy);
+			endpoint.addInterceptor(new ConcurrencyInterceptor(concurrencyPolicy));
 		}
 		return endpoint;
 	}

org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java

@@ -28,6 +28,7 @@ import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.endpoint.TargetEndpoint;
+import org.springframework.integration.endpoint.interceptor.ConcurrencyInterceptor;
 import org.springframework.integration.handler.MethodInvokingTarget;
 import org.springframework.integration.message.MessageTarget;
 import org.springframework.integration.scheduling.Subscription;
@@ -70,7 +71,7 @@ public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostP
 					concurrencyAnnotation.maxSize());
 			concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds());
 			concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity());
-			endpoint.setConcurrencyPolicy(concurrencyPolicy);
+			endpoint.addInterceptor(new ConcurrencyInterceptor(concurrencyPolicy));
 		}
 		return endpoint;
 	}

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

@@ -25,7 +25,6 @@
 				</xsd:documentation>
 			</xsd:annotation>
 			<xsd:sequence>
-				<xsd:element name="default-concurrency" type="concurrencyType" minOccurs="0" maxOccurs="1"/>
 				<xsd:element ref="interceptor" minOccurs="0" maxOccurs="unbounded"/>
 			</xsd:sequence>
 			<xsd:attribute name="auto-startup" type="xsd:boolean"/>
@@ -304,12 +303,13 @@
 		</xsd:complexType>
 	</xsd:element>
 
-	<xsd:complexType name="concurrencyType">
+	<xsd:complexType name="concurrencyInterceptorType">
 		<xsd:annotation>
 			<xsd:documentation>
-	Defines a concurrency policy.
+	Defines a ConcurrencyInterceptor for endpoints.
 			</xsd:documentation>
 		</xsd:annotation>
+		<xsd:attribute name="task-executor" type="xsd:string"/>
 		<xsd:attribute name="core" type="xsd:int"/>
 		<xsd:attribute name="max" type="xsd:int"/>
 		<xsd:attribute name="queue-capacity" type="xsd:int"/>
@@ -415,11 +415,9 @@
 			<xsd:extension base="beans:identifiedType">
 				<xsd:all>
 					<xsd:element ref="schedule" minOccurs="0" maxOccurs="1"/>
-					<xsd:element name="concurrency" type="concurrencyType" minOccurs="0" maxOccurs="1"/>
 					<xsd:element name="interceptors" type="interceptorsType" minOccurs="0" maxOccurs="1"/>
 				</xsd:all>
 				<xsd:attribute name="input-channel" type="xsd:string" use="required"/>
-				<xsd:attribute name="error-handler" type="xsd:string"/>
 				<xsd:attribute name="selector" type="xsd:string"/>
 			</xsd:extension>
 		</xsd:complexContent>
@@ -453,6 +451,7 @@
 					</xsd:complexType>
 				</xsd:element>
 				<xsd:element name="transaction-interceptor" type="transactionInterceptorType" minOccurs="0" maxOccurs="1"/>
+				<xsd:element name="concurrency-interceptor" type="concurrencyInterceptorType" minOccurs="0" maxOccurs="1"/>
 				<xsd:any namespace="##other" processContents="strict" minOccurs="0" maxOccurs="unbounded"/>				
 			</xsd:choice>
 		</xsd:sequence>

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java

@@ -59,18 +59,23 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware
 		return (this.name != null) ? this.name : super.toString();
 	}
 
+	public void addInterceptor(Object interceptor) {
+		if (interceptor instanceof Advice) {
+			this.interceptors.add((Advice) interceptor);
+		}
+		else if (interceptor instanceof EndpointInterceptor) {
+			this.interceptors.add(new EndpointMethodInterceptor((EndpointInterceptor) interceptor));
+		}
+		else {
+			throw new ConfigurationException("Interceptor must implement either "
+					+ "'" + Advice.class.getName() + "' or '" + EndpointInterceptor.class.getName() + "'.");
+		}
+	}
+
 	public void setInterceptors(List<Object> interceptors) {
+		this.interceptors.clear();
 		for (Object interceptor : interceptors) {
-			if (interceptor instanceof Advice) {
-				this.interceptors.add((Advice) interceptor);
-			}
-			else if (interceptor instanceof EndpointInterceptor) {
-				this.interceptors.add(new EndpointMethodInterceptor((EndpointInterceptor) interceptor));
-			}
-			else {
-				throw new ConfigurationException("Each interceptor element must implement either "
-						+ "'" + Advice.class.getName() + "' or '" + EndpointInterceptor.class.getName() + "'.");
-			}
+			this.addInterceptor(interceptor);
 		}
 	}
 

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ConcurrentTarget.java

@@ -1,108 +0,0 @@
-/*
- * Copyright 2002-2008 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.integration.endpoint;
-
-import java.util.concurrent.ExecutorService;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.springframework.beans.factory.DisposableBean;
-import org.springframework.integration.handler.MessageHandlerNotRunningException;
-import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
-import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageDeliveryException;
-import org.springframework.integration.message.MessageTarget;
-import org.springframework.integration.util.ErrorHandler;
-import org.springframework.util.Assert;
-
-/**
- * A {@link MessageTarget} implementation that encapsulates an Executor and delegates
- * to a wrapped target for concurrent, asynchronous message handling.
- * 
- * @author Mark Fisher
- */
-public class ConcurrentTarget implements MessageTarget, DisposableBean {
-
-	private final Log logger = LogFactory.getLog(this.getClass());
-
-	private final MessageTarget target;
-
-	private final ExecutorService executor;
-
-	private volatile ErrorHandler errorHandler;
-
-
-	public ConcurrentTarget(MessageTarget target, ExecutorService executor) {
-		Assert.notNull(target, "'target' must not be null");
-		Assert.notNull(executor, "'executor' must not be null");
-		this.target = target;
-		this.executor = executor;
-	}
-
-
-	public void setErrorHandler(ErrorHandler errorHandler) {
-		this.errorHandler = errorHandler;
-	}
-
-	public void destroy() {
-		this.executor.shutdown();
-	}
-
-	public boolean send(Message<?> message) {
-		if (this.executor.isShutdown()) {
-			throw new MessageHandlerNotRunningException(message);
-		}
-		try {
-			this.executor.execute(new TargetTask(message));
-			return true;
-		}
-		catch (RuntimeException e) {
-			throw new MessageHandlerRejectedExecutionException(message, e);
-		}
-	}
-
-
-	private class TargetTask implements Runnable {
-
-		private Message<?> message;
-
-		TargetTask(Message<?> message) {
-			this.message = message;
-		}
-
-		public void run() {
-			try {
-				if (!target.send(this.message)) {
-					throw new MessageDeliveryException(message, "failed to send message to target");
-				}
-			}
-			catch (Throwable t) {
-				if (logger.isDebugEnabled()) {
-					logger.debug("error occurred in handler execution", t);
-				}
-				if (errorHandler != null) {
-					errorHandler.handle(t);
-				}
-				else if (logger.isWarnEnabled() && !logger.isDebugEnabled()) {
-					logger.warn("error occurred in handler execution", t);
-				}
-			}
-		}
-	}
-
-}

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java

@@ -35,7 +35,7 @@ import org.springframework.util.StringUtils;
  * 
  * @author Mark Fisher
  */
-public class HandlerEndpoint extends TargetEndpoint {
+public class HandlerEndpoint extends TargetEndpoint implements MessageProducingEndpoint {
 
 	private volatile MessageHandler handler;
 
@@ -81,10 +81,6 @@ public class HandlerEndpoint extends TargetEndpoint {
 		this.outputChannelName = outputChannelName;
 	}
 
-	public String getOutputChannelName() {
-		return this.outputChannelName;
-	}
-
 	public void setReturnAddressOverrides(boolean returnAddressOverrides) {
 		this.returnAddressOverrides = returnAddressOverrides;
 	}
@@ -132,7 +128,7 @@ public class HandlerEndpoint extends TargetEndpoint {
 		return null;
 	}
 
-	private MessageChannel getOutputChannel() {
+	public MessageChannel getOutputChannel() {
 		ChannelRegistry registry = this.getChannelRegistry();
 		if (this.outputChannelName != null && registry != null) {
 			return registry.lookupChannel(this.outputChannelName);
@@ -140,6 +136,10 @@ public class HandlerEndpoint extends TargetEndpoint {
 		return null;
 	}
 
+	public String getOutputChannelName() {
+		return this.outputChannelName;
+	}
+
 
 	private static class HandlerInvokingTarget implements MessageTarget {
 

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageConsumingEndpoint.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.endpoint;
+
+import org.springframework.integration.scheduling.Subscription;
+
+/**
+ * @author Mark Fisher
+ */
+public interface MessageConsumingEndpoint {
+
+	Subscription getSubscription();
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/MessageProducingEndpoint.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.endpoint;
+
+import org.springframework.integration.channel.MessageChannel;
+
+/**
+ * @author Mark Fisher
+ */
+public interface MessageProducingEndpoint {
+
+	String getOutputChannelName();
+
+	MessageChannel getOutputChannel();
+
+}

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java

@@ -16,26 +16,16 @@
 
 package org.springframework.integration.endpoint;
 
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.ChannelRegistry;
 import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.handler.MessageHandlerNotRunningException;
-import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageHandlingException;
 import org.springframework.integration.message.MessageTarget;
 import org.springframework.integration.message.selector.MessageSelector;
 import org.springframework.integration.scheduling.Subscription;
-import org.springframework.integration.util.ErrorHandler;
 import org.springframework.util.Assert;
 
 /**
@@ -43,16 +33,13 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class TargetEndpoint extends AbstractEndpoint implements MessageTarget, ChannelRegistryAware, InitializingBean, Lifecycle {
+public class TargetEndpoint extends AbstractEndpoint
+		implements MessageTarget, MessageConsumingEndpoint, ChannelRegistryAware, InitializingBean, Lifecycle {
 
 	private volatile MessageTarget target;
 
 	private volatile Subscription subscription;
 
-	private volatile ConcurrencyPolicy concurrencyPolicy;
-
-	private volatile ErrorHandler errorHandler;
-
 	private volatile MessageSelector selector;
 
 	private volatile ChannelRegistry channelRegistry;
@@ -92,22 +79,6 @@ public class TargetEndpoint extends AbstractEndpoint implements MessageTarget, C
 		this.subscription = subscription;
 	}
 
-	public ConcurrencyPolicy getConcurrencyPolicy() {
-		return this.concurrencyPolicy;
-	}
-
-	public void setConcurrencyPolicy(ConcurrencyPolicy concurrencyPolicy) {
-		this.concurrencyPolicy = concurrencyPolicy;
-	}
-
-	public void setErrorHandler(ErrorHandler errorHandler) {
-		this.errorHandler = errorHandler;
-	}
-
-	public boolean hasErrorHandler() {
-		return (this.errorHandler != null);
-	}
-
 	/**
 	 * Set the channel registry to use for looking up channels by name.
 	 */
@@ -123,18 +94,6 @@ public class TargetEndpoint extends AbstractEndpoint implements MessageTarget, C
 		if (this.target instanceof ChannelRegistryAware && this.channelRegistry != null) {
 			((ChannelRegistryAware) this.target).setChannelRegistry(this.channelRegistry);
 		}
-		if (this.concurrencyPolicy != null && !(this.target instanceof ConcurrentTarget)) {
-			int capacity = this.concurrencyPolicy.getQueueCapacity();
-			BlockingQueue<Runnable> queue = (capacity < 1) ? new SynchronousQueue<Runnable>() : new ArrayBlockingQueue<Runnable>(capacity);
-			ExecutorService executor = new ThreadPoolExecutor(this.concurrencyPolicy.getCoreSize(), this.concurrencyPolicy.getMaxSize(),
-					this.concurrencyPolicy.getKeepAliveSeconds(), TimeUnit.SECONDS, queue);
-			this.target = new ConcurrentTarget(this.target, executor);
-		}
-		if (this.target instanceof ConcurrentTarget) {
-			if (this.errorHandler != null) {
-				((ConcurrentTarget) this.target).setErrorHandler(this.errorHandler);
-			}
-		}
 		this.initialized = true;
 	}
 
@@ -180,23 +139,7 @@ public class TargetEndpoint extends AbstractEndpoint implements MessageTarget, C
 		if (this.selector != null && !this.selector.accept(message)) {
 			return false;
 		}
-		try {
-			return this.target.send(message);
-		}
-		catch (MessageHandlerRejectedExecutionException e) {
-			throw e;
-		}
-		catch (Throwable t) {
-			if (this.errorHandler == null) {
-				if (t instanceof MessageHandlingException) {
-					throw (MessageHandlingException) t;
-				}
-				throw new MessageHandlingException(message,
-						"error occurred in endpoint, and no 'errorHandler' available", t);
-			}
-			this.errorHandler.handle(t);
-			return false;
-		}
+		return this.target.send(message);
 	}
 
 	@Override

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/interceptor/ConcurrencyInterceptor.java

@@ -0,0 +1,137 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.endpoint.interceptor;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.aopalliance.intercept.MethodInvocation;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.Lifecycle;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.integration.bus.MessageBus;
+import org.springframework.integration.bus.MessageBusAware;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.endpoint.ConcurrencyPolicy;
+import org.springframework.integration.endpoint.EndpointInterceptor;
+import org.springframework.integration.handler.MessageHandlerNotRunningException;
+import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
+import org.springframework.integration.util.ErrorHandler;
+import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
+import org.springframework.util.Assert;
+
+/**
+ * An {@link EndpointInterceptor} implementation that delegates to a
+ * {@link TaskExecutor} for concurrent, asynchronous message handling.
+ * 
+ * @author Mark Fisher
+ */
+public class ConcurrencyInterceptor extends EndpointInterceptorAdapter
+		implements DisposableBean, InitializingBean, MessageBusAware {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private final TaskExecutor executor;
+
+	private volatile ErrorHandler errorHandler;
+
+	private volatile MessageBus messageBus;
+
+
+	public ConcurrencyInterceptor(TaskExecutor executor) {
+		Assert.notNull(executor, "TaskExecutor must not be null");
+		this.executor = executor;
+	}
+
+	public ConcurrencyInterceptor(ConcurrencyPolicy concurrencyPolicy) {
+		Assert.notNull(concurrencyPolicy, "ConcurrencyPolicy must not be null");
+		int core = concurrencyPolicy.getCoreSize();
+		int max = concurrencyPolicy.getMaxSize();
+		int keepAlive = concurrencyPolicy.getKeepAliveSeconds();
+		int capacity = concurrencyPolicy.getQueueCapacity();
+		BlockingQueue<Runnable> queue = (capacity > 0) ?
+				new LinkedBlockingQueue<Runnable>(capacity) : new SynchronousQueue<Runnable>();
+		ThreadPoolExecutor tpe = new ThreadPoolExecutor(core, max, keepAlive, TimeUnit.SECONDS, queue);
+		this.executor = new ConcurrentTaskExecutor(tpe);
+	}
+
+
+	public void setErrorHandler(ErrorHandler errorHandler) {
+		this.errorHandler = errorHandler;
+	}
+
+	public void setMessageBus(MessageBus messageBus) {
+		this.messageBus = messageBus;
+	}
+
+	public void afterPropertiesSet() throws Exception {
+		if (this.errorHandler == null) {
+			MessageChannel errorChannel = this.messageBus.getErrorChannel();
+			if (errorChannel != null) {
+				this.errorHandler = new MessagePublishingErrorHandler(errorChannel);
+			}
+		}
+	}
+
+	public void destroy() throws Exception {
+		if (this.executor instanceof DisposableBean) {
+			((DisposableBean) this.executor).destroy();
+		}
+	}
+
+	@Override
+	public boolean aroundInvoke(final MethodInvocation invocation) {
+		final Message<?> message = (Message<?>) invocation.getArguments()[0];
+		if (invocation.getThis() instanceof Lifecycle && !((Lifecycle) invocation.getThis()).isRunning()) {
+			throw new MessageHandlerNotRunningException(message);
+		}
+		try {
+			this.executor.execute(new Runnable() {
+				public void run() {
+					try {
+						invocation.proceed();
+					}
+					catch (Throwable t) {
+						if (logger.isDebugEnabled()) {
+							logger.debug("error occurred in handler execution", t);
+						}
+						if (errorHandler != null) {
+							errorHandler.handle(t);
+						}
+						else if (logger.isWarnEnabled() && !logger.isDebugEnabled()) {
+							logger.warn("error occurred in handler execution", t);
+						}
+					}
+				}
+			});
+			return true;
+		}
+		catch (RuntimeException e) {
+			throw new MessageHandlerRejectedExecutionException(message, e);
+		}
+	}
+
+}