Added support for interceptors on MessageEndpoints.

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

@@ -27,6 +27,7 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -55,8 +56,11 @@ import org.springframework.integration.endpoint.MessagingGateway;
 import org.springframework.integration.endpoint.SourceEndpoint;
 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.Target;
 import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
+import org.springframework.integration.scheduling.MessagingTask;
 import org.springframework.integration.scheduling.MessagingTaskScheduler;
 import org.springframework.integration.scheduling.Schedule;
 import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
@@ -408,20 +412,27 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 		}
 	}
 
-	private void registerSourceEndpoint(String name, SourceEndpoint endpoint) {
+	private void registerSourceEndpoint(String name, final SourceEndpoint endpoint) {
 		if (!this.initialized) {
 			this.initialize();
 		}
-		this.taskScheduler.schedule(endpoint);
+		final Schedule schedule = endpoint.getSchedule();
+		if (schedule != null) {
+			this.taskScheduler.schedule(new MessagingTask() {
+				public Schedule getSchedule() {
+					return schedule;
+				}
+				public void run() {
+					endpoint.invoke(new CommandMessage(new PollCommand()));
+				}
+			});
+		}
 		if (endpoint instanceof Lifecycle) {
 			this.lifecycleEndpoints.add((Lifecycle) endpoint);
 			if (this.isRunning()) {
 				((Lifecycle) endpoint).start();
 			}
 		}
-		if (logger.isInfoEnabled()) {
-			logger.info("registered source adapter '" + name + "'");
-		}
 	}
 
 	private void registerGateway(String name, MessagingGateway gateway) {
@@ -436,7 +447,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 		}
 	}
 
-	private void activateSubscription(MessageChannel channel, Target target, Schedule schedule) {
+	private void activateSubscription(MessageChannel channel, TargetEndpoint targetEndpoint, Schedule schedule) {
 		SubscriptionManager manager = this.subscriptionManagers.get(channel);
 		if (manager == null) {
 			if (logger.isWarnEnabled()) {
@@ -445,7 +456,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 			}
 			return;
 		}
-		manager.addTarget(target, schedule);
+		manager.addTarget(targetEndpoint, schedule);
 		if (this.isRunning() && !manager.isRunning()) {
 			manager.start();
 		}

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

@@ -28,7 +28,6 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.dispatcher.DefaultPollingDispatcher;
 import org.springframework.integration.dispatcher.DirectChannel;
 import org.springframework.integration.dispatcher.PollingDispatcherTask;
 import org.springframework.integration.endpoint.TargetEndpoint;
@@ -122,10 +121,9 @@ public class SubscriptionManager {
 		}
 		PollingDispatcherTask dispatcherTask = this.dispatcherTasks.get(schedule);
 		if (dispatcherTask == null) {
-			DefaultPollingDispatcher dispatcher = new DefaultPollingDispatcher(this.channel);
-			dispatcherTask = this.dispatcherTasks.putIfAbsent(schedule, new PollingDispatcherTask(dispatcher, schedule));
+			dispatcherTask = this.dispatcherTasks.putIfAbsent(schedule, new PollingDispatcherTask(this.channel, schedule));
 		}
-		this.dispatcherTasks.get(schedule).getDispatcher().subscribe(target);
+		this.dispatcherTasks.get(schedule).subscribe(target);
 		if (dispatcherTask == null && this.isRunning()) {
 			this.scheduleDispatcherTask(schedule);
 		}
@@ -135,7 +133,7 @@ public class SubscriptionManager {
 		boolean removed = false;
 		Collection<PollingDispatcherTask> dispatcherTaskValues = this.dispatcherTasks.values();
 		for (PollingDispatcherTask dispatcherTask : dispatcherTaskValues) {
-			removed = (removed || dispatcherTask.getDispatcher().unsubscribe(target));
+			removed = (removed || dispatcherTask.unsubscribe(target));
 		}
 		return removed;
 	}

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

@@ -24,6 +24,7 @@ import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.parsing.BeanComponentDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
@@ -61,6 +62,8 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef
 
 	private static final String CONCURRENCY_POLICY_PROPERTY = "concurrencyPolicy";
 
+	private static final String ADVICE_CHAIN_ELEMENT = "advice-chain";
+
 
 	@Override
 	protected boolean shouldGenerateId() {
@@ -98,6 +101,10 @@ public abstract class AbstractTargetEndpointParser extends AbstractSingleBeanDef
 				else if (SCHEDULE_ELEMENT.equals(localName)) {
 					schedule = this.parseSchedule((Element) child);
 				}
+				else if (ADVICE_CHAIN_ELEMENT.equals(localName)) {
+					ManagedList adviceChain = IntegrationNamespaceUtils.parseEndpointAdviceChain((Element) child);
+					builder.addPropertyValue("adviceChain", adviceChain);
+				}
 			}
 		}
 		if (StringUtils.hasText(inputChannel)) {

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

@@ -17,8 +17,11 @@
 package org.springframework.integration.config;
 
 import org.w3c.dom.Element;
+import org.w3c.dom.Node;
+import org.w3c.dom.NodeList;
 
 import org.springframework.beans.factory.config.RuntimeBeanReference;
+import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.util.StringUtils;
@@ -61,6 +64,23 @@ public abstract class IntegrationNamespaceUtils {
 		return policy;
 	}
 
+	@SuppressWarnings("unchecked")
+	public static ManagedList parseEndpointAdviceChain(Element element) {
+		ManagedList adviceChain = new ManagedList();
+		NodeList childNodes = element.getChildNodes();
+		for (int i = 0; i < childNodes.getLength(); i++) {
+			Node child = childNodes.item(i);
+			if (child.getNodeType() == Node.ELEMENT_NODE) {
+				String localName = child.getLocalName();
+				if ("ref".equals(localName)) {
+					String ref = ((Element) child).getAttribute("bean");
+					adviceChain.add(new RuntimeBeanReference(ref));
+				}
+			}
+		}
+		return adviceChain;
+	}
+
 	/**
 	 * 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/MessageBusParser.java

@@ -16,13 +16,17 @@
 
 package org.springframework.integration.config;
 
+import org.w3c.dom.Element;
+import org.w3c.dom.Node;
+import org.w3c.dom.NodeList;
+
 import org.springframework.beans.factory.BeanDefinitionStoreException;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.RuntimeBeanReference;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.support.ManagedList;
-import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.core.Conventions;
@@ -30,9 +34,6 @@ import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.bus.MessageBusAwareBeanPostProcessor;
 import org.springframework.util.StringUtils;
-import org.w3c.dom.Element;
-import org.w3c.dom.Node;
-import org.w3c.dom.NodeList;
 
 /**
  * Parser for the <em>message-bus</em> element of the integration namespace.
@@ -100,6 +101,7 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
 		this.processChildElements(beanDefinition, element);
 	}
 
+	@SuppressWarnings("unchecked")
 	private void processChildElements(BeanDefinitionBuilder beanDefinition, Element element) {
 		NodeList childNodes = element.getChildNodes();
 		ManagedList interceptors = new ManagedList();
@@ -131,11 +133,21 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
 	 * Adds extra post-processors to the context, to inject the objects configured by the MessageBus
 	 */
 	private void addPostProcessors(ParserContext parserContext) {
-		BeanDefinition postProcessorDefinition = new RootBeanDefinition(MessageBusAwareBeanPostProcessor.class);
-		postProcessorDefinition.getConstructorArgumentValues().addGenericArgumentValue(
-				new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
-		parserContext.getRegistry().registerBeanDefinition(MESSAGE_BUS_AWARE_POST_PROCESSOR_BEAN_NAME,
-				postProcessorDefinition);
+		this.registerMessageBusAwarePostProcessor(parserContext);
+		this.registerMessageEndpointPostProcessor(parserContext);
+	}
+
+	private void registerMessageBusAwarePostProcessor(ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MessageBusAwareBeanPostProcessor.class);
+		builder.addConstructorArgReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
+		builder.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
+		parserContext.getRegistry().registerBeanDefinition(MESSAGE_BUS_AWARE_POST_PROCESSOR_BEAN_NAME, builder.getBeanDefinition());
+	}
+
+	private void registerMessageEndpointPostProcessor(ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MessageEndpointBeanPostProcessor.class);
+		builder.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
+		BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
 	}
 
 }

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

@@ -0,0 +1,75 @@
+/*
+ * 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.config;
+
+import java.lang.reflect.Method;
+import java.util.List;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.support.StaticMethodMatcherPointcutAdvisor;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.message.Message;
+
+/**
+ * A post-processor that applies an advice-chain by creating a proxy for an endpoint.
+ *
+ * @author Mark Fisher
+ */
+public class MessageEndpointBeanPostProcessor implements BeanPostProcessor {
+
+	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+		return bean;
+	}
+
+	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+		if (bean instanceof AbstractEndpoint) {
+			AbstractEndpoint endpoint = (AbstractEndpoint) bean;
+			List<Advice> adviceChain = endpoint.getAdviceChain();
+			if (adviceChain.size() > 0) {
+				ProxyFactory proxyFactory = new ProxyFactory(endpoint);
+				for (Advice advice : adviceChain) {
+					proxyFactory.addAdvisor(new EndpointInvokeMethodAdvisor(advice));
+				}
+				bean = proxyFactory.getProxy();
+			}
+		}
+		return bean;
+	}
+
+
+	@SuppressWarnings("serial")
+	private static class EndpointInvokeMethodAdvisor extends StaticMethodMatcherPointcutAdvisor {
+
+		EndpointInvokeMethodAdvisor(Advice advice) {
+			super(advice);
+		}
+
+
+		@SuppressWarnings("unchecked")
+		public boolean matches(Method method, Class clazz) {
+			return method.getName().equals("invoke")
+					&& method.getParameterTypes().length == 1
+					&& method.getParameterTypes()[0].equals(Message.class);
+		}
+
+	}
+
+}

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

@@ -19,6 +19,7 @@ package org.springframework.integration.config;
 import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.support.ManagedList;
 import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.ConfigurationException;
@@ -67,7 +68,12 @@ public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser {
 		if (scheduleElement == null) {
 			throw new ConfigurationException("The <schedule/> sub-element is required for a <source-endpoint/>.");
 		}
-		builder.addConstructorArgValue(this.parseSchedule(scheduleElement));
+		builder.addPropertyValue("schedule", this.parseSchedule(scheduleElement));
+		Element adviceChainElement = DomUtils.getChildElementByTagName(element, "advice-chain");
+		if (adviceChainElement != null) {
+			ManagedList adviceChain = IntegrationNamespaceUtils.parseEndpointAdviceChain(adviceChainElement);
+			builder.addPropertyValue("adviceChain", adviceChain);
+		}
 	}
 
 	/**

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

@@ -76,7 +76,9 @@ public class SourceAnnotationPostProcessor extends AbstractAnnotationMethodPostP
 			outputChannel = new DirectChannel();
 			this.getMessageBus().registerChannel(beanName + ".output", outputChannel);
 		}
-		return new SourceEndpoint((Source<?>) bean, outputChannel, schedule);
+		SourceEndpoint endpoint = new SourceEndpoint((Source<?>) bean, outputChannel);
+		endpoint.setSchedule(schedule);
+		return endpoint;
 	}
 
 }

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

@@ -166,6 +166,7 @@
 				<xsd:extension base="beans:identifiedType">
 					<xsd:sequence>
 						<xsd:element ref="schedule" minOccurs="1" maxOccurs="1"/>
+						<xsd:element name="advice-chain" type="adviceChainType" minOccurs="0" maxOccurs="1"/>
 					</xsd:sequence>
 					<xsd:attribute name="source" type="xsd:string" use="required"/>
 					<xsd:attribute name="channel" type="xsd:string" use="required"/>
@@ -415,6 +416,7 @@
 				<xsd:all>
 					<xsd:element ref="schedule" minOccurs="0" maxOccurs="1"/>
 					<xsd:element name="concurrency" type="concurrencyType" minOccurs="0" maxOccurs="1"/>
+					<xsd:element name="advice-chain" type="adviceChainType" minOccurs="0" maxOccurs="1"/>
 				</xsd:all>
 				<xsd:attribute name="input-channel" type="xsd:string" use="required"/>
 				<xsd:attribute name="error-handler" type="xsd:string"/>
@@ -437,4 +439,19 @@
 		</xsd:complexContent>
 	</xsd:complexType>
 
+	<xsd:complexType name="adviceChainType">
+		<xsd:annotation>
+			<xsd:documentation>
+	Defines a list of Advice.
+			</xsd:documentation>
+		</xsd:annotation>
+		<xsd:sequence>
+			<xsd:element name="ref" minOccurs="0" maxOccurs="unbounded">
+				<xsd:complexType>
+					<xsd:attribute name="bean" use="required"/>
+				</xsd:complexType>
+			</xsd:element>
+		</xsd:sequence>
+	</xsd:complexType>
+
 </xsd:schema>

org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/DefaultPollingDispatcher.java

@@ -1,97 +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.dispatcher;
-
-import java.util.LinkedList;
-import java.util.List;
-
-import org.springframework.integration.channel.DispatcherPolicy;
-import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.BlockingSource;
-import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageDeliveryAware;
-import org.springframework.integration.message.MessageDeliveryException;
-import org.springframework.integration.message.Source;
-import org.springframework.util.Assert;
-
-/**
- * A subclass of {@link SimpleDispatcher} that adds message retrieval
- * capabilities by polling a {@link PollableSource} for {@link Message Messages}.
- * The number of messages retrieved per poll is limited by the '<em>maxMessagesPerTask</em>'
- * property of the provided {@link DispatcherPolicy}, and the timeout for each
- * receive call is determined by the policy's '<em>receiveTimeout</em>'
- * property. In general, it is recommended to use a value of 1 (the default) for
- * 'maxMessagesPerTask' whenever a significant timeout is provided. Otherwise
- * the poller may be holding on to available messages while waiting for
- * additional messages. Note that the 'timeout' value is only relevant if the
- * specified source is an implementation of {@link BlockingSource}. The default
- * timeout value is 0 indicating that the method should return immediately
- * rather than waiting for a {@link Message} to become available.
- * 
- * @author Mark Fisher
- */
-public class DefaultPollingDispatcher extends SimpleDispatcher implements PollingDispatcher {
-
-	private final Source<?> source;
-
-
-	public DefaultPollingDispatcher(MessageChannel channel) {
-		this(channel, channel.getDispatcherPolicy());
-	}
-
-	public DefaultPollingDispatcher(Source<?> source, DispatcherPolicy dispatcherPolicy) {
-		super(dispatcherPolicy);
-		Assert.notNull(source, "source must not be null");
-		this.source = source;
-		this.dispatcherPolicy.setReceiveTimeout(0);
-	}
-
-	@Override
-	public boolean dispatch(Message<?> message) {
-		boolean sent = super.dispatch(message);
-		if (this.source instanceof MessageDeliveryAware) {
-			if (sent) {
-				((MessageDeliveryAware) this.source).onSend(message);
-			}
-			else {
-				((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message"));
-			}
-		}
-		return sent;
-	}
-
-	public List<Message<?>> poll() {
-		List<Message<?>> messages = new LinkedList<Message<?>>();
-		int limit = this.dispatcherPolicy.getMaxMessagesPerTask();
-		while (messages.size() < limit) {
-			Message<?> message = null;
-			long timeout = this.dispatcherPolicy.getReceiveTimeout();
-			if (this.source instanceof BlockingSource && timeout >= 0) {
-				message = ((BlockingSource<?>) this.source).receive(timeout);
-			}
-			else {
-				message = this.source.receive();
-			}
-			if (message == null) {
-				return messages;
-			}
-			messages.add(message);
-		}
-		return messages;
-	}
-
-}

org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java

@@ -16,9 +16,10 @@
 
 package org.springframework.integration.dispatcher;
 
-import java.util.List;
-
+import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.Subscribable;
+import org.springframework.integration.message.Target;
 import org.springframework.integration.scheduling.MessagingTask;
 import org.springframework.integration.scheduling.Schedule;
 import org.springframework.util.Assert;
@@ -28,22 +29,29 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class PollingDispatcherTask implements MessagingTask {
+public class PollingDispatcherTask implements MessagingTask, Subscribable {
 
-	private final PollingDispatcher dispatcher;
+	private final MessageChannel channel;
 
 	private final Schedule schedule;
 
+	private final SimpleDispatcher dispatcher;
 
-	public PollingDispatcherTask(PollingDispatcher dispatcher, Schedule schedule) {
-		Assert.notNull(dispatcher, "dispatcher must not be null");
-		this.dispatcher = dispatcher;
+
+	public PollingDispatcherTask(MessageChannel channel, Schedule schedule) {
+		Assert.notNull(channel, "channel must not be null");
+		this.channel = channel;
 		this.schedule = schedule;
+		this.dispatcher = new SimpleDispatcher(this.channel.getDispatcherPolicy());
 	}
 
 
-	public PollingDispatcher getDispatcher() {
-		return this.dispatcher;
+	public boolean subscribe(Target target) {
+		return this.dispatcher.subscribe(target);
+	}
+
+	public boolean unsubscribe(Target target) {
+		return this.dispatcher.unsubscribe(target);
 	}
 
 	public Schedule getSchedule() {
@@ -51,9 +59,16 @@ public class PollingDispatcherTask implements MessagingTask {
 	}
 
 	public void run() {
-		List<Message<?>> messages = this.dispatcher.poll();
-		for (Message<?> message : messages) {
+		long timeout = this.channel.getDispatcherPolicy().getReceiveTimeout();
+		int limit = this.channel.getDispatcherPolicy().getMaxMessagesPerTask();
+		int count = 0;
+		while (count < limit) {
+			Message<?> message = (timeout < 0) ? this.channel.receive() : this.channel.receive(timeout);
+			if (message == null) {
+				return;
+			}
 			this.dispatcher.dispatch(message);
+			count++;
 		}
 	}
 

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

@@ -16,20 +16,32 @@
 
 package org.springframework.integration.endpoint;
 
+import java.util.ArrayList;
+import java.util.List;
+
+import org.aopalliance.aop.Advice;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.integration.ConfigurationException;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageHandlingException;
+
 /**
  * Base class for {@link MessageEndpoint} implementations.
  * 
  * @author Mark Fisher
  */
-public abstract class AbstractEndpoint implements MessageEndpoint {
+public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware {
 
 	protected final Log logger = LogFactory.getLog(this.getClass());
 
 	private volatile String name;
 
+	private final List<Advice> adviceChain = new ArrayList<Advice>();
+
 
 	public String getName() {
 		return this.name;
@@ -47,4 +59,37 @@ public abstract class AbstractEndpoint implements MessageEndpoint {
 		return (this.name != null) ? this.name : super.toString();
 	}
 
+	public void setAdviceChain(List<Object> adviceChain) {
+		for (Object advice : adviceChain) {
+			if (advice instanceof Advice) {
+				this.adviceChain.add((Advice) advice);
+			}
+			else if (advice instanceof EndpointInterceptor) {
+				this.adviceChain.add(new EndpointMethodInterceptor((EndpointInterceptor) advice));
+			}
+			else {
+				throw new ConfigurationException("Each adviceChain element must implement either "
+						+ "'" + Advice.class.getName() + "' or '" + EndpointInterceptor.class.getName() + "'.");
+			}
+		}
+	}
+
+	public List<Advice> getAdviceChain() {
+		return this.adviceChain;
+	}
+
+	public final boolean invoke(Message<?> message) {
+		if (message == null) {
+			throw new IllegalArgumentException("Message must not be null.");
+		}
+		if (!this.supports(message)) {
+			throw new MessageHandlingException(message, "unsupported message");
+		}
+		return this.doInvoke(message);
+	}
+
+	protected abstract boolean supports(Message<?> message);
+
+	protected abstract boolean doInvoke(Message<?> message);
+
 }

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

@@ -0,0 +1,34 @@
+/*
+ * 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.aopalliance.intercept.MethodInvocation;
+
+import org.springframework.integration.message.Message;
+
+/**
+ * @author Mark Fisher
+ */
+public interface EndpointInterceptor {
+
+	boolean preInvoke(Message message);
+
+	boolean aroundInvoke(MethodInvocation invocation) throws Throwable;
+
+	void postInvoke(Message message, boolean result);
+
+}

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

@@ -0,0 +1,39 @@
+/*
+ * 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.aopalliance.intercept.MethodInvocation;
+
+import org.springframework.integration.message.Message;
+
+/**
+ * @author Mark Fisher
+ */
+public class EndpointInterceptorAdapter implements EndpointInterceptor {
+
+	public boolean preInvoke(Message message) {
+		return true;
+	}
+
+	public boolean aroundInvoke(MethodInvocation invocation) throws Throwable {
+		return (Boolean) invocation.proceed();
+	}
+
+	public void postInvoke(Message message, boolean result) {
+	}
+
+}

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

@@ -0,0 +1,56 @@
+/*
+ * 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.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+
+import org.springframework.integration.message.Message;
+import org.springframework.util.Assert;
+
+/**
+ * @author Mark Fisher
+ */
+public class EndpointMethodInterceptor implements MethodInterceptor {
+
+	private final EndpointInterceptor interceptor;
+
+
+	public EndpointMethodInterceptor(EndpointInterceptor interceptor) {
+		Assert.notNull(interceptor, "EndpointInterceptor must not be null.");
+		this.interceptor = interceptor;
+	}
+
+
+	public Object invoke(MethodInvocation invocation) throws Throwable {
+		Message<?> message = null;
+		try {
+			message = (Message<?>) invocation.getArguments()[0];
+		}
+		catch (Exception e) {
+			throw new IllegalStateException("EndpointMethodInterceptor is only applicable for the "
+					+ "'MessageEndpoint.invoke(Message message)' method.");
+		}
+		if (!this.interceptor.preInvoke(message)) {
+			return Boolean.FALSE;
+		}
+		boolean returnValue = this.interceptor.aroundInvoke(invocation);
+		this.interceptor.postInvoke(message, returnValue);
+		return returnValue;
+	}
+
+}

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

@@ -16,15 +16,17 @@
 
 package org.springframework.integration.endpoint;
 
-import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.integration.message.Message;
 
 /**
  * Base interface for message endpoints.
  * 
  * @author Mark Fisher
  */
-public interface MessageEndpoint extends BeanNameAware {
+public interface MessageEndpoint {
 
 	String getName();
 
+	boolean invoke(Message<?> message);
+
 }

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

@@ -16,21 +16,14 @@
 
 package org.springframework.integration.endpoint;
 
-import java.lang.reflect.Method;
-import java.util.List;
-
-import org.aopalliance.aop.Advice;
-
-import org.springframework.aop.framework.ProxyFactory;
-import org.springframework.aop.support.StaticMethodMatcherPointcutAdvisor;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.channel.DispatcherPolicy;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.dispatcher.DefaultPollingDispatcher;
-import org.springframework.integration.dispatcher.PollingDispatcher;
-import org.springframework.integration.dispatcher.PollingDispatcherTask;
+import org.springframework.integration.dispatcher.SimpleDispatcher;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageDeliveryAware;
+import org.springframework.integration.message.MessageDeliveryException;
+import org.springframework.integration.message.PollCommand;
 import org.springframework.integration.message.Source;
-import org.springframework.integration.scheduling.MessagingTask;
 import org.springframework.integration.scheduling.Schedule;
 import org.springframework.util.Assert;
 
@@ -40,120 +33,50 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class SourceEndpoint extends AbstractEndpoint implements MessagingTask, InitializingBean {
+public class SourceEndpoint extends AbstractEndpoint {
 
-	private final Schedule schedule;
+	private final Source<?> source;
 
-	private final DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
+	private final SimpleDispatcher dispatcher = new SimpleDispatcher(new DispatcherPolicy());
 
-	private volatile PollingDispatcher dispatcher;
-
-	private volatile List<Advice> dispatchAdviceChain;
-
-	private volatile MessagingTask task;
-
-	private volatile List<Advice> taskAdviceChain;
-
-	private volatile boolean taskInitialized;
-
-	private final Object taskMonitor = new Object();
+	private volatile Schedule schedule;
 
 
-	public SourceEndpoint(Source<?> source, MessageChannel channel, Schedule schedule) {
+	public SourceEndpoint(Source<?> source, MessageChannel channel) {
 		Assert.notNull(source, "source must not be null");
 		Assert.notNull(channel, "channel must not be null");
-		Assert.notNull(schedule, "schedule must not be null");
-		this.dispatcher = new DefaultPollingDispatcher(source, this.dispatcherPolicy);
+		this.source = source;
 		this.dispatcher.subscribe(channel);
+	}
+
+
+	public void setSchedule(Schedule schedule) {
 		this.schedule = schedule;
 	}
 
-
-	public void setMaxMessagesPerTask(int maxMessagesPerTask) {
-		this.dispatcherPolicy.setMaxMessagesPerTask(maxMessagesPerTask);
-	}
-
-	public void setSendTimeout(long sendTimeout) {
-		this.dispatcher.setSendTimeout(sendTimeout);
-	}
-
-	public void setTaskAdviceChain(List<Advice> taskAdviceChain) {
-		this.taskAdviceChain = taskAdviceChain;
-	}
-
-	public void setDispatchAdviceChain(List<Advice> dispatchAdviceChain) {
-		this.dispatchAdviceChain = dispatchAdviceChain;
-	}
-
 	public Schedule getSchedule() {
 		return this.schedule;
 	}
 
-	public void afterPropertiesSet() {
-		this.initializeTask();
+	protected boolean supports(Message<?> message) {
+		return (message.getPayload() instanceof PollCommand);
 	}
 
-	public void initializeTask() {
-		synchronized (this.taskMonitor) {
-			if (this.taskInitialized) {
-				return;
-			}
-			this.refreshTask();
-			this.taskInitialized = true;
+	public final boolean doInvoke(Message<?> pollCommandMessage) {
+		Message<?> message = this.source.receive();
+		if (message == null) {
+			return false;
 		}
-	}
-
-	public void refreshTask() {
-		synchronized (this.taskMonitor) {
-			PollingDispatcher dispatcherProxy = null;
-			if (this.dispatchAdviceChain != null && this.dispatchAdviceChain.size() > 0) {
-				ProxyFactory proxyFactory = new ProxyFactory(this.dispatcher);
-				proxyFactory.setInterfaces(new Class[] { PollingDispatcher.class });
-				for (Advice advice : this.dispatchAdviceChain) {
-					proxyFactory.addAdvisor(new MethodNameAdvisor(advice, "dispatch"));
-				}
-				dispatcherProxy = (PollingDispatcher) proxyFactory.getProxy();
+		boolean sent = this.dispatcher.dispatch(message);
+		if (this.source instanceof MessageDeliveryAware) {
+			if (sent) {
+				((MessageDeliveryAware) this.source).onSend(message);
 			}
-			this.task = new PollingDispatcherTask((dispatcherProxy != null) ? dispatcherProxy : this.dispatcher, this.schedule);
-			if (this.taskAdviceChain != null && this.taskAdviceChain.size() > 0) {
-				ProxyFactory proxyFactory = new ProxyFactory(this.task);
-				proxyFactory.setInterfaces(new Class[] { MessagingTask.class });
-				for (Advice advice : this.taskAdviceChain) {
-					proxyFactory.addAdvisor(new MethodNameAdvisor(advice, "run"));
-				}
-				this.task = (MessagingTask) proxyFactory.getProxy();
+			else {
+				((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message"));
 			}
 		}
-	}
-
-	private MessagingTask getTask() {
-		synchronized (this.taskMonitor) {
-			if (!this.taskInitialized) {
-				this.initializeTask();
-			}
-			return this.task;
-		}
-	}
-
-	public void run() {
-		this.getTask().run();
-	}
-
-
-	@SuppressWarnings("serial")
-	private static class MethodNameAdvisor extends StaticMethodMatcherPointcutAdvisor {
-
-		private final String methodName;
-
-		MethodNameAdvisor(Advice advice, String methodName) {
-			super(advice);
-			this.methodName = methodName;
-		}
-
-		@SuppressWarnings("unchecked")
-		public boolean matches(Method method, Class targetClass) {
-			return method.getName().equals(methodName);
-		}
+		return sent;
 	}
 
 }

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

@@ -169,7 +169,8 @@ public class TargetEndpoint extends AbstractEndpoint implements Target, ChannelR
 		this.running = false;
 	}
 
-	public final boolean send(Message<?> message) {
+	@Override
+	protected final boolean doInvoke(Message<?> message) {
 		if (logger.isDebugEnabled()) {
 			logger.debug("endpoint '" + this + "' handling message: " + message);
 		}
@@ -198,4 +199,13 @@ public class TargetEndpoint extends AbstractEndpoint implements Target, ChannelR
 		}
 	}
 
+	@Override
+	protected final boolean supports(Message<?> message) {
+		return true;
+	}
+
+	public boolean send(Message<?> message) {
+		return this.invoke(message);
+	}
+
 }

org.springframework.integration/src/main/java/org/springframework/integration/message/Command.java

@@ -16,13 +16,11 @@
 
 package org.springframework.integration.message;
 
-import java.util.List;
-
 /**
+ * A marker interface for commands.
+ * 
  * @author Mark Fisher
  */
-public interface Poller {
-
-	 List<Message<?>> poll();
+public interface Command {
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/message/CommandMessage.java

@@ -14,14 +14,15 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.dispatcher;
-
-import org.springframework.integration.message.Poller;
-import org.springframework.integration.message.Subscribable;
+package org.springframework.integration.message;
 
 /**
  * @author Mark Fisher
  */
-public interface PollingDispatcher extends Poller, MessageDispatcher, Subscribable {
+public class CommandMessage extends GenericMessage<Command> {
+
+	public CommandMessage(Command command) {
+		super(command);
+	}
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/message/PollCommand.java

@@ -0,0 +1,24 @@
+/*
+ * 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.message;
+
+/**
+ * @author Mark Fisher
+ */
+public class PollCommand implements Command {
+
+}