Added AbstractMessageConsumingEndpoint. MessageDispatchers now expect MessageConsumer instances as subscribers, and the MessageEndpoint no longer has a send() method or a getSource() method. All consumer endpoints now use 'inputChannel' as the property (instead of source). The MessageBus is less involved in endpoint activation now, since endpoints that need to poll a channel can create, configure, and schedule their own poller.

org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierEndpoint.java

@@ -151,7 +151,8 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractInOutEndpoi
 	 * Initialize this endpoint.
 	 */
 	@Override
-	protected void initialize() {
+	protected void initialize()  throws Exception {
+		super.initialize();
 		this.trackedCorrelationIds = new ArrayBlockingQueue<Object>(this.trackedCorrelationIdCapacity);
 		this.executor.scheduleWithFixedDelay(new ReaperTask(),
 				this.reaperInterval, this.reaperInterval, TimeUnit.MILLISECONDS);

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

@@ -44,18 +44,12 @@ import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.DefaultChannelRegistry;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
-import org.springframework.integration.channel.PollableChannel;
-import org.springframework.integration.endpoint.AbstractPoller;
-import org.springframework.integration.endpoint.ChannelPoller;
 import org.springframework.integration.endpoint.DefaultEndpointRegistry;
 import org.springframework.integration.endpoint.EndpointRegistry;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.endpoint.MessagingGateway;
-import org.springframework.integration.message.MessageSource;
-import org.springframework.integration.message.SubscribableSource;
-import org.springframework.integration.scheduling.PollingSchedule;
-import org.springframework.integration.scheduling.Schedule;
 import org.springframework.integration.scheduling.TaskScheduler;
+import org.springframework.integration.scheduling.TaskSchedulerAware;
 import org.springframework.integration.scheduling.spi.ProviderTaskScheduler;
 import org.springframework.integration.scheduling.spi.SimpleScheduleServiceProvider;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
@@ -78,14 +72,10 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 
 	private final EndpointRegistry endpointRegistry = new DefaultEndpointRegistry();
 
-	private final Set<AbstractPoller> pollers = new CopyOnWriteArraySet<AbstractPoller>();
-
-	private volatile Schedule defaultPollerSchedule = new PollingSchedule(0);
-
-	private final List<Lifecycle> lifecycleEndpoints = new CopyOnWriteArrayList<Lifecycle>();
-
 	private final MessageBusInterceptorsList interceptors = new MessageBusInterceptorsList();
 
+	private final Set<Lifecycle> lifecycleGateways = new CopyOnWriteArraySet<Lifecycle>();
+
 	private volatile TaskScheduler taskScheduler;
 
 	private volatile ApplicationContext applicationContext;
@@ -263,39 +253,46 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 		}
 	}
 
+	private void deactivateEndpoints() {
+		Set<String> endpointNames = this.endpointRegistry.getEndpointNames();
+		for (String name : endpointNames) {
+			MessageEndpoint endpoint = this.endpointRegistry.lookupEndpoint(name);
+			if (endpoint != null) {
+				this.deactivateEndpoint(endpoint);
+			}
+		}
+	}
+
 	private void activateEndpoint(MessageEndpoint endpoint) {
 		Assert.notNull(endpoint, "'endpoint' must not be null");
 		if (endpoint instanceof ChannelRegistryAware) {
 			((ChannelRegistryAware) endpoint).setChannelRegistry(this);
 		}
-		MessageSource<?> source = endpoint.getSource();
-		if (source == null) {
-			throw new ConfigurationException("endpoint '" + endpoint + "' has no source");
+		if (endpoint instanceof TaskSchedulerAware) {
+			((TaskSchedulerAware) endpoint).setTaskScheduler(this.taskScheduler);
 		}
-		if (source instanceof SubscribableSource) {
-			((SubscribableSource) source).subscribe(endpoint);
-			if (source instanceof AbstractPoller) {
-				AbstractPoller poller = (AbstractPoller) source;
-				this.pollers.add(poller);
-				this.taskScheduler.schedule(poller);
-			}
-			return;
-		}
-		else if (source instanceof PollableChannel) {
-			ChannelPoller poller = new ChannelPoller((PollableChannel) source, this.defaultPollerSchedule);
-			poller.subscribe(endpoint);
-			this.pollers.add(poller);
-			this.taskScheduler.schedule(poller);
+		if (endpoint instanceof Lifecycle) {
+			((Lifecycle) endpoint).start();
 		}
 		if (logger.isInfoEnabled()) {
-			logger.info("activated subscription to channel '"
-					+ source + "' for endpoint '" + endpoint + "'");
+			logger.info("activated endpoint '" + endpoint + "'");
 		}
 	}
 
+	public void deactivateEndpoint(MessageEndpoint endpoint) {
+		Assert.notNull(endpoint, "'endpoint' must not be null");
+		if (endpoint instanceof Lifecycle) {
+			((Lifecycle) endpoint).stop();
+			if (this.logger.isInfoEnabled()) {
+				logger.info("deactivated endpoint '" + endpoint + "'");
+			}
+		}
+	}
+
+	// TODO: once gateways are endpoints, remove this 
 	private void registerGateway(String name, MessagingGateway gateway) {
 		if (gateway instanceof Lifecycle) {
-			this.lifecycleEndpoints.add((Lifecycle) gateway);
+			this.lifecycleGateways.add((Lifecycle) gateway);
 			if (this.isRunning()) {
 				((Lifecycle) gateway).start();
 			}
@@ -305,19 +302,6 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 		}
 	}
 
-	public void deactivateEndpoint(MessageEndpoint endpoint) {
-		Assert.notNull(endpoint, "'endpoint' must not be null");
-		for (AbstractPoller poller : this.pollers) {
-			boolean removed = ((AbstractPoller) poller).unsubscribe(endpoint);
-			if (removed && this.logger.isInfoEnabled()) {
-				logger.info("unsubscribed endpoint '" + endpoint + "' from poller '" + poller + "'");
-			}
-		}
-		if (endpoint instanceof Lifecycle) {
-			((Lifecycle) endpoint).stop();
-		}
-	}
-
 	public boolean isRunning() {
 		synchronized (this.lifecycleMonitor) {
 			return this.running;
@@ -335,14 +319,11 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 		this.starting = true;
 		synchronized (this.lifecycleMonitor) {
 			this.activateEndpoints();
+			for (Lifecycle gateway : this.lifecycleGateways) {
+				gateway.start();
+			}
 			this.taskScheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel()));
 			this.taskScheduler.start();
-			for (Lifecycle endpoint : this.lifecycleEndpoints) {
-				endpoint.start();
-				if (logger.isInfoEnabled()) {
-					logger.info("started endpoint '" + endpoint + "'");
-				}
-			}
 		}
 		this.running = true;
 		this.starting = false;
@@ -358,14 +339,12 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
 		}
 		this.interceptors.preStop();
 		synchronized (this.lifecycleMonitor) {
+			this.deactivateEndpoints();
+			for (Lifecycle gateway : this.lifecycleGateways) {
+				gateway.stop();
+			}
 			this.running = false;
 			this.taskScheduler.stop();
-			for (Lifecycle endpoint : this.lifecycleEndpoints) {
-				endpoint.stop();
-				if (logger.isInfoEnabled()) {
-					logger.info("stopped endpoint '" + endpoint + "'");
-				}
-			}
 		}
 		this.interceptors.postStop();
 		if (logger.isInfoEnabled()) {

org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java

@@ -19,6 +19,7 @@ package org.springframework.integration.channel;
 import org.springframework.integration.dispatcher.SimpleDispatcher;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.message.SubscribableSource;
 
 /**
@@ -33,12 +34,12 @@ public class DirectChannel extends AbstractMessageChannel implements Subscribabl
 	private final SimpleDispatcher dispatcher = new SimpleDispatcher();
 
 
-	public boolean subscribe(MessageEndpoint endpoint) {
-		return this.dispatcher.subscribe(endpoint);
+	public boolean subscribe(MessageConsumer consumer) {
+		return this.dispatcher.subscribe(consumer);
 	}
 
-	public boolean unsubscribe(MessageEndpoint endpoint) {
-		return this.dispatcher.unsubscribe(endpoint);
+	public boolean unsubscribe(MessageConsumer consumer) {
+		return this.dispatcher.unsubscribe(consumer);
 	}
 
 	@Override

org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java

@@ -18,8 +18,8 @@ package org.springframework.integration.channel;
 
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.dispatcher.BroadcastingDispatcher;
-import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.message.SubscribableSource;
 
 /**
@@ -48,12 +48,12 @@ public class PublishSubscribeChannel extends AbstractMessageChannel implements S
 		this.dispatcher.setApplySequence(applySequence);
 	}
 
-	public boolean subscribe(MessageEndpoint endpoint) {
-		return this.dispatcher.subscribe(endpoint);
+	public boolean subscribe(MessageConsumer consumer) {
+		return this.dispatcher.subscribe(consumer);
 	}
 
-	public boolean unsubscribe(MessageEndpoint endpoint) {
-		return this.dispatcher.unsubscribe(endpoint);
+	public boolean unsubscribe(MessageConsumer consumer) {
+		return this.dispatcher.unsubscribe(consumer);
 	}
 
 	@Override

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

@@ -89,12 +89,13 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio
 		}
 		Element pollerElement = DomUtils.getChildElementByTagName(element, POLLER_ELEMENT);
 		if (pollerElement != null) {
-			String pollerBeanName = IntegrationNamespaceUtils.parseChannelPoller(inputChannel, pollerElement, parserContext);
-			builder.addPropertyReference("source", pollerBeanName);
-		}
-		else {
-			builder.addPropertyReference("source", inputChannel);
+			IntegrationNamespaceUtils.configureSchedule(pollerElement, builder);
+			Element txElement = DomUtils.getChildElementByTagName(pollerElement, "transactional");
+			if (txElement != null) {
+				IntegrationNamespaceUtils.configureTransactionAttributes(txElement, builder);
+			}
 		}
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, INPUT_CHANNEL_ATTRIBUTE);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, SELECTOR_ATTRIBUTE);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, ERROR_HANDLER_ATTRIBUTE);

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

@@ -73,19 +73,21 @@ public class ChannelAdapterParser extends AbstractBeanDefinitionParser {
 				source = BeanDefinitionReaderUtils.registerWithGeneratedName(invokerBuilder.getBeanDefinition(), parserContext.getRegistry());
 			}
 			adapterBuilder =  BeanDefinitionBuilder.genericBeanDefinition(InboundChannelAdapter.class);
-			if (pollerElement != null) {
-				String pollerBeanName = IntegrationNamespaceUtils.parseSourcePoller(source, pollerElement, parserContext);
-				adapterBuilder.addPropertyReference("source", pollerBeanName);
-			}
-			else {
-				adapterBuilder.addPropertyReference("source", source);
-			}
+			adapterBuilder.addPropertyReference("source", source);
 			if (StringUtils.hasText(channelName)) {
 				adapterBuilder.addPropertyReference("channel", channelName);
 			}
 			else {
 				adapterBuilder.addPropertyReference("channel", this.createDirectChannel(element, parserContext));
 			}
+			if (pollerElement != null) {
+				IntegrationNamespaceUtils.configureSchedule(pollerElement, adapterBuilder);
+				IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, pollerElement, "max-messages-per-poll");
+				Element txElement = DomUtils.getChildElementByTagName(pollerElement, "transactional");
+				if (txElement != null) {
+					IntegrationNamespaceUtils.configureTransactionAttributes(txElement, adapterBuilder);
+				}
+			}
 		}
 		else if (StringUtils.hasText(target)) {
 			if (StringUtils.hasText(methodName)) {
@@ -100,14 +102,17 @@ public class ChannelAdapterParser extends AbstractBeanDefinitionParser {
 				if (!StringUtils.hasText(channelName)) {
 					throw new ConfigurationException("outbound channel-adapter with a 'poller' requires a 'channel' to poll");
 				}
-				String pollerBeanName = IntegrationNamespaceUtils.parseChannelPoller(channelName, pollerElement, parserContext);
-				adapterBuilder.addPropertyReference("source", pollerBeanName);
+				IntegrationNamespaceUtils.configureSchedule(pollerElement, adapterBuilder);
+				Element txElement = DomUtils.getChildElementByTagName(pollerElement, "transactional");
+				if (txElement != null) {
+					IntegrationNamespaceUtils.configureTransactionAttributes(txElement, adapterBuilder);
+				}
 			}
-			else if (StringUtils.hasText(channelName)) {
-				adapterBuilder.addPropertyReference("source", channelName);
+			if (StringUtils.hasText(channelName)) {
+				adapterBuilder.addPropertyReference("inputChannel", channelName);
 			}
 			else {
-				adapterBuilder.addPropertyReference("source",
+				adapterBuilder.addPropertyReference("inputChannel",
 						this.createDirectChannel(element, parserContext));
 			}
 		}

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

@@ -21,19 +21,15 @@ import org.w3c.dom.Element;
 import org.springframework.beans.factory.config.BeanDefinitionHolder;
 import org.springframework.beans.factory.parsing.BeanComponentDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.core.Conventions;
 import org.springframework.integration.ConfigurationException;
-import org.springframework.integration.endpoint.ChannelPoller;
-import org.springframework.integration.endpoint.SourcePoller;
 import org.springframework.integration.scheduling.CronSchedule;
 import org.springframework.integration.scheduling.PollingSchedule;
 import org.springframework.integration.scheduling.Schedule;
 import org.springframework.transaction.support.DefaultTransactionDefinition;
 import org.springframework.util.StringUtils;
-import org.springframework.util.xml.DomUtils;
 
 /**
  * Shared utility methods for integration namespace parsers.
@@ -138,71 +134,54 @@ public abstract class IntegrationNamespaceUtils {
 	}
 
 	/**
-	 * Parse a "poller" element to create a ChannelPoller and return the bean name of the poller instance.
+	 * Parse a "poller" element to create a Schedule and add it to the property values of the target builder.
 	 * 
-	 * @param channelBeanName the name of the PollableChannel bean
-	 * @param element the "poller" element to parse
-	 * @param parserContext the parserContext for registering a newly created bean definition
-	 * @return the name of the ChannelPoller bean definition
+	 * @param pollerElement the "poller" element to parse
+	 * @param targetBuilder the builder that expects the "schedule" property
 	 */
-	public static String parseChannelPoller(String channelBeanName, Element element, ParserContext parserContext) {
-		return parsePoller(channelBeanName, element, parserContext, true);
-	}
-
-	/**
-	 * Parse a "poller" element to create a SourcePoller and return the bean name of the poller instance.
-	 * 
-	 * @param sourceBeanName the name of the PollableSource bean
-	 * @param element the "poller" element to parse
-	 * @param parserContext the parserContext for registering a newly created bean definition
-	 * @return the name of the poller bean definition
-	 */
-	public static String parseSourcePoller(String sourceBeanName, Element element, ParserContext parserContext) {
-		return parsePoller(sourceBeanName, element, parserContext, false);
-	}
-
-	private static String parsePoller(String sourceBeanName, Element element, ParserContext parserContext, boolean isChannel) {
-		Class<?> beanClass = isChannel ? ChannelPoller.class : SourcePoller.class;
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(beanClass);
+	public static void configureSchedule(Element pollerElement, BeanDefinitionBuilder targetBuilder) {
 		Schedule schedule = null;
-		if (!(StringUtils.hasText(element.getAttribute("period")) ^ StringUtils.hasText(element.getAttribute("cron")))) {
+		if (!(StringUtils.hasText(pollerElement.getAttribute("period")) ^ StringUtils.hasText(pollerElement.getAttribute("cron")))) {
 			throw new ConfigurationException("A <poller> element must define either a period "
 					+ "or a cron expression (but not both)");
 		}
-		if (StringUtils.hasText(element.getAttribute("period"))) {
-			Long period = Long.valueOf(element.getAttribute("period"));
+		if (StringUtils.hasText(pollerElement.getAttribute("period"))) {
+			Long period = Long.valueOf(pollerElement.getAttribute("period"));
 			schedule = new PollingSchedule(period);
-			String initialDelay = element.getAttribute("initial-delay");
+			String initialDelay = pollerElement.getAttribute("initial-delay");
 			if (StringUtils.hasText(initialDelay)) {
 				((PollingSchedule)schedule).setInitialDelay(Long.valueOf(initialDelay));
 			}
-			if ("true".equals(element.getAttribute("fixed-rate").toLowerCase())) {
+			if ("true".equals(pollerElement.getAttribute("fixed-rate").toLowerCase())) {
 				((PollingSchedule)schedule).setFixedRate(true);
 			}
 			else {
 				((PollingSchedule)schedule).setFixedRate(false);
 			}
 		}
-		if (StringUtils.hasText(element.getAttribute("cron"))) {
-		   schedule = new CronSchedule(element.getAttribute("cron"));
+		if (StringUtils.hasText(pollerElement.getAttribute("cron"))) {
+		   schedule = new CronSchedule(pollerElement.getAttribute("cron"));
 		}
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
-		Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
-		if (txElement != null) {
-			builder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager"));
-			builder.addPropertyValue("propagationBehaviorName",
-					DefaultTransactionDefinition.PREFIX_PROPAGATION + txElement.getAttribute("propagation"));
-			builder.addPropertyValue("isolationLevelName",
-					DefaultTransactionDefinition.PREFIX_ISOLATION + txElement.getAttribute("isolation"));
-			builder.addPropertyValue("transactionTimeout", txElement.getAttribute("timeout"));
-			builder.addPropertyValue("transactionReadOnly", txElement.getAttribute("read-only"));
-		}
-		builder.addConstructorArgReference(sourceBeanName);
-		builder.addConstructorArgValue(schedule);
-		setValueIfAttributeDefined(builder, element, "receive-timeout");
-		setValueIfAttributeDefined(builder, element, "send-timeout");
-		setValueIfAttributeDefined(builder, element, "max-messages-per-poll");
-		return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
+		targetBuilder.addPropertyValue("schedule", schedule);
+	}
+
+	/**
+	 * Parse a "transactional" element and configure the "transactionManager" and "transactionDefinition"
+	 * properties for the target builder.
+	 * 
+	 * @param txElement the "transactional" element to parse
+	 * @param targetBuilder the builder that expects the "transactionManager" and "transactionDefinition" properties
+	 */
+	public static void configureTransactionAttributes(Element txElement, BeanDefinitionBuilder targetBuilder) {
+		targetBuilder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager"));
+		DefaultTransactionDefinition txDefinition = new DefaultTransactionDefinition();
+		txDefinition.setPropagationBehaviorName(
+				DefaultTransactionDefinition.PREFIX_PROPAGATION + txElement.getAttribute("propagation"));
+		txDefinition.setIsolationLevelName(
+				DefaultTransactionDefinition.PREFIX_ISOLATION + txElement.getAttribute("isolation"));
+		txDefinition.setTimeout(Integer.valueOf(txElement.getAttribute("timeout")));
+		txDefinition.setReadOnly(txElement.getAttribute("read-only").equalsIgnoreCase("true"));
+		targetBuilder.addPropertyValue("transactionDefinition", txDefinition);
 	}
 
 }

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

@@ -51,7 +51,7 @@ public class ResequencerParser extends AbstractSimpleBeanDefinitionParser {
 
 	@Override
 	protected void postProcess(BeanDefinitionBuilder builder, Element element) {
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, INPUT_CHANNEL_ATTRIBUTE, "source");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, INPUT_CHANNEL_ATTRIBUTE);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
 	}

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

@@ -28,7 +28,7 @@ import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.endpoint.AbstractInOutEndpoint;
-import org.springframework.integration.endpoint.ChannelPoller;
+import org.springframework.integration.endpoint.AbstractMessageConsumingEndpoint;
 import org.springframework.integration.scheduling.PollingSchedule;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -87,22 +87,22 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 			if (inputChannel == null) {
 				throw new ConfigurationException("unable to resolve inputChannel '" + inputChannelName + "'");
 			}
-			if (pollerAnnotation != null) {
-				if (inputChannel instanceof PollableChannel) {
-					PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());
-					schedule.setInitialDelay(pollerAnnotation.initialDelay());
-					schedule.setFixedRate(pollerAnnotation.fixedRate());
-					schedule.setTimeUnit(pollerAnnotation.timeUnit());
-					ChannelPoller poller = new ChannelPoller((PollableChannel) inputChannel, schedule);
-					poller.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
-					endpoint.setSource(poller);
+			if (endpoint instanceof AbstractMessageConsumingEndpoint) {
+				AbstractMessageConsumingEndpoint consumingEndpoint = (AbstractMessageConsumingEndpoint) endpoint;
+				if (pollerAnnotation != null) {	
+					if (inputChannel instanceof PollableChannel) {
+						PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());
+						schedule.setInitialDelay(pollerAnnotation.initialDelay());
+						schedule.setFixedRate(pollerAnnotation.fixedRate());
+						schedule.setTimeUnit(pollerAnnotation.timeUnit());
+						consumingEndpoint.setSchedule(schedule);
+						consumingEndpoint.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
+					}
+					else {
+						throw new ConfigurationException("The @Poller annotation should only be provided for a PollableChannel");
+					}
 				}
-				else {
-					throw new ConfigurationException("The @Poller annotation should only be provided for a PollableSource");
-				}
-			}
-			else {
-				endpoint.setSource(inputChannel);
+				consumingEndpoint.setInputChannel(inputChannel);
 			}
 			if (endpoint instanceof AbstractInOutEndpoint) {
 				String outputChannelName = (String) AnnotationUtils.getValue(annotation, OUTPUT_CHANNEL_ATTRIBUTE);

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

@@ -26,11 +26,9 @@ import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PollableChannel;
-import org.springframework.integration.endpoint.ChannelPoller;
 import org.springframework.integration.endpoint.InboundChannelAdapter;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.endpoint.OutboundChannelAdapter;
-import org.springframework.integration.endpoint.SourcePoller;
 import org.springframework.integration.handler.MethodInvokingTarget;
 import org.springframework.integration.message.MethodInvokingSource;
 import org.springframework.integration.scheduling.PollingSchedule;
@@ -90,32 +88,22 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
 					+ "when using the @ChannelAdapter annotation with a no-arg method.");
 		}
 		Schedule schedule = this.createSchedule(pollerAnnotation);
-		SourcePoller poller = new SourcePoller(source, schedule);
-		int maxMessagesPerPoll = pollerAnnotation.maxMessagesPerPoll();
-		if (maxMessagesPerPoll == -1) {
-			// the default is 1 since a MethodInvokingSource might return a non-null value
-			// every time it is invoked, thus producing an infinite number of messages per poll
-			maxMessagesPerPoll = 1;
-		}
-		poller.setMaxMessagesPerPoll(maxMessagesPerPoll);
 		InboundChannelAdapter adapter = new InboundChannelAdapter();
-		adapter.setSource(poller);
+		adapter.setSource(source);
 		adapter.setChannel(channel);
+		adapter.setSchedule(schedule);
 		adapter.setBeanName(this.generateUniqueName(channel.getName() + ".inboundAdapter"));
 		return adapter;
 	}
 
 	private OutboundChannelAdapter createOutboundChannelAdapter(MethodInvokingTarget target, MessageChannel channel, Poller pollerAnnotation) {
 		OutboundChannelAdapter adapter = new OutboundChannelAdapter(target);
+		adapter.setInputChannel(channel);
 		if (channel instanceof PollableChannel) {
 			Schedule schedule = (pollerAnnotation != null)
 					? this.createSchedule(pollerAnnotation)
 					: new PollingSchedule(0);
-			ChannelPoller poller = new ChannelPoller((PollableChannel) channel, schedule);
-			adapter.setSource(poller);
-		}
-		else {
-			adapter.setSource(channel);
+			adapter.setSchedule(schedule);
 		}
 		adapter.setBeanName(this.generateUniqueName(channel.getName() + ".outboundAdapter"));
 		return adapter;

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

@@ -23,8 +23,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.core.task.TaskExecutor;
-import org.springframework.integration.endpoint.MessageEndpoint;
-import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageConsumer;
 
 /**
  * Base class for {@link MessageDispatcher} implementations.
@@ -35,21 +34,21 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
 
 	protected final Log logger = LogFactory.getLog(this.getClass());
 
-	protected final Set<MessageEndpoint> endpoints = new CopyOnWriteArraySet<MessageEndpoint>();
+	protected final Set<MessageConsumer> subscribers = new CopyOnWriteArraySet<MessageConsumer>();
 
 	private volatile TaskExecutor taskExecutor;
 
 
-	public boolean subscribe(MessageEndpoint endpoint) {
-		return this.endpoints.add(endpoint);
+	public boolean subscribe(MessageConsumer consumer) {
+		return this.subscribers.add(consumer);
 	}
 
-	public boolean unsubscribe(MessageEndpoint endpoint) {
-		return this.endpoints.remove(endpoint);
+	public boolean unsubscribe(MessageConsumer consumer) {
+		return this.subscribers.remove(consumer);
 	}
 
 	/**
-	 * Specify a {@link TaskExecutor} for invoking the endpoints.
+	 * Specify a {@link TaskExecutor} for invoking the consumers.
 	 * If none is provided, the invocation will occur in the thread
 	 * that runs this polling dispatcher.
 	 */
@@ -61,15 +60,8 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
 		return this.taskExecutor;
 	}
 
-	/**
-	 * A convenience method for subclasses to send a Message to a single endpoint.
-	 */
-	protected final boolean sendMessageToEndpoint(Message<?> message, MessageEndpoint endpoint) {
-		return endpoint.send(message);
-	}
-
 	public String toString() {
-		return this.getClass().getSimpleName() + " with endpoints: " + this.endpoints;
+		return this.getClass().getSimpleName() + " with subscribers: " + this.subscribers;
 	}
 
 }

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

@@ -17,9 +17,9 @@
 package org.springframework.integration.dispatcher;
 
 import org.springframework.core.task.TaskExecutor;
-import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.message.MessageConsumer;
 
 /**
  * A broadcasting dispatcher implementation. It makes a best effort to
@@ -45,8 +45,8 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
 
 	public boolean dispatch(Message<?> message) {
 		int sequenceNumber = 1;
-		int sequenceSize = this.endpoints.size();
-		for (final MessageEndpoint endpoint : this.endpoints) {
+		int sequenceSize = this.subscribers.size();
+		for (final MessageConsumer consumer : this.subscribers) {
 			final Message<?> messageToSend = (!this.applySequence) ? message
 				: MessageBuilder.fromMessage(message)
 						.setSequenceNumber(sequenceNumber++)
@@ -56,12 +56,12 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
 			if (executor != null) {
 				executor.execute(new Runnable() {
 					public void run() {
-						sendMessageToEndpoint(messageToSend, endpoint);
+						consumer.onMessage(messageToSend);
 					}
 				});
 			}
 			else {
-				this.sendMessageToEndpoint(messageToSend, endpoint);
+				consumer.onMessage(messageToSend);
 			}
 		}
 		return true;

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

@@ -16,50 +16,45 @@
 
 package org.springframework.integration.dispatcher;
 
-import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.message.MessageDeliveryException;
 import org.springframework.integration.message.MessageRejectedException;
 
 /**
  * Basic implementation of {@link MessageDispatcher} that will attempt
- * to send a {@link Message} to one of its endpoints. As soon as <em>one</em>
- * of the endpoints accepts the Message, the dispatcher will return 'true'.
+ * to send a {@link Message} to one of its subscribers. As soon as <em>one</em>
+ * of the subscribers accepts the Message, the dispatcher will return 'true'.
  * <p>
- * If the dispatcher has no endpoints, a {@link MessageDeliveryException}
- * will be thrown. If all endpoints reject the Message, the dispatcher will
- * throw a MessageRejectedException. If all endpoints return 'false'
- * (e.g. due to a timeout), the dispatcher will return 'false'.
+ * If the dispatcher has no subscribers, a {@link MessageDeliveryException}
+ * will be thrown. If all subscribers reject the Message, the dispatcher will
+ * throw a MessageRejectedException.
  * 
  * @author Mark Fisher
  */
 public class SimpleDispatcher extends AbstractDispatcher {
 
 	public boolean dispatch(Message<?> message) {
-		if (this.endpoints.size() == 0) {
+		if (this.subscribers.size() == 0) {
 			throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
 		}
 		int count = 0;
 		int rejectedExceptionCount = 0;
-		for (MessageEndpoint endpoint : this.endpoints) {
+		for (MessageConsumer consumer : this.subscribers) {
 			count++;
 			try {
-				if (this.sendMessageToEndpoint(message, endpoint)) {
-					return true;
-				}
-				if (logger.isDebugEnabled()) {
-					logger.debug("Failed to send message to endpoint, continuing with other endpoints if available.");
-				}
+				consumer.onMessage(message);
+				return true;
 			}
 			catch (MessageRejectedException e) {
 				rejectedExceptionCount++;
 				if (logger.isDebugEnabled()) {
-					logger.debug("Endpoint '" + endpoint + "' rejected Message, continuing with other endpoints if available.", e);
+					logger.debug("Consumer '" + consumer + "' rejected Message, continuing with other subscribers if available.", e);
 				}
 			}
 		}
 		if (rejectedExceptionCount == count) {
-			throw new MessageRejectedException(message, "All of dispatcher's endpoints rejected Message.");
+			throw new MessageRejectedException(message, "All of dispatcher's subscribers rejected Message.");
 		}
 		return false;
 	}

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

@@ -24,31 +24,35 @@ import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.channel.ChannelRegistry;
 import org.springframework.integration.channel.ChannelRegistryAware;
-import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageExchangeTemplate;
-import org.springframework.integration.message.MessageHandlingException;
-import org.springframework.integration.message.MessageSource;
 import org.springframework.integration.message.MessagingException;
-import org.springframework.integration.message.SubscribableSource;
+import org.springframework.integration.scheduling.TaskScheduler;
+import org.springframework.integration.scheduling.TaskSchedulerAware;
 import org.springframework.integration.util.ErrorHandler;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
 
 /**
  * The base class for Message Endpoint implementations.
  * 
  * @author Mark Fisher
  */
-public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegistryAware, BeanNameAware, InitializingBean {
+public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegistryAware, TaskSchedulerAware, BeanNameAware, InitializingBean {
 
 	protected final Log logger = LogFactory.getLog(this.getClass());
 
 	private volatile String name;
 
-	private MessageSource<?> source;
+	private volatile ChannelRegistry channelRegistry;
+
+	private volatile TaskScheduler taskScheduler;
+
+	private volatile PlatformTransactionManager transactionManager;
+
+	private volatile TransactionDefinition transactionDefinition;
 
 	private volatile ErrorHandler errorHandler;
 
-	private volatile ChannelRegistry channelRegistry;
-
 	private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate();
 
 
@@ -63,14 +67,6 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
 		this.name = name;
 	}
 
-	public MessageSource<?> getSource() {
-		return this.source;
-	}
-
-	public void setSource(MessageSource<?> source) {
-		this.source = source;
-	}
-
 	protected ChannelRegistry getChannelRegistry() {
 		return this.channelRegistry;
 	}
@@ -79,6 +75,22 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
 		this.channelRegistry = channelRegistry;
 	}
 
+	protected TaskScheduler getTaskScheduler() {
+		return this.taskScheduler;
+	}
+
+	public void setTaskScheduler(TaskScheduler taskScheduler) {
+		this.taskScheduler = taskScheduler;
+	}
+
+	public void setTransactionManager(PlatformTransactionManager transactionManager) {
+		this.transactionManager = transactionManager;
+	}
+
+	public void setTransactionDefinition(TransactionDefinition transactionDefinition) {
+		this.transactionDefinition= transactionDefinition;
+	}
+
 	protected MessageExchangeTemplate getMessageExchangeTemplate() {
 		return this.messageExchangeTemplate;
 	}
@@ -94,9 +106,6 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
 	}
 
 	public final void afterPropertiesSet() {
-		if (this.source != null && (this.source instanceof SubscribableSource)) {
-			((SubscribableSource) this.source).subscribe(this);
-		}
 		try {
 			this.initialize();
 		}
@@ -114,31 +123,7 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
 	protected void initialize()  throws Exception {
 	}
 
-	public final boolean send(Message<?> message) {
-		if (message == null || message.getPayload() == null) {
-			throw new IllegalArgumentException("Message and its payload must not be null");
-		}
-		if (this.logger.isDebugEnabled()) {
-			this.logger.debug("endpoint '" + this + "' processing message: " + message);
-		}
-		try {
-			return this.sendInternal(message);
-		}
-		catch (Exception e) {
-			if (e instanceof MessagingException) {
-				this.handleException((MessagingException) e);
-			}
-			else {
-				this.handleException(new MessageHandlingException(message,
-						"failure occurred in endpoint '" + this.toString() + "'", e));
-			}
-			return false;
-		}
-	}
-
-	protected abstract boolean sendInternal(Message<?> message);
-
-	private void handleException(MessagingException exception) {
+	protected void handleException(MessagingException exception) {
 		if (this.errorHandler == null) {
 			if (this.logger.isWarnEnabled()) {
 				this.logger.warn("exception occurred in endpoint '" + this.name + "'", exception);
@@ -148,6 +133,18 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist
 		this.errorHandler.handle(exception);
 	}
 
+	protected final void configureTransactionSettingsForPoller(AbstractPoller poller) {
+		if (this.transactionManager != null) {
+			poller.setTransactionManager(this.transactionManager);
+		}
+		if (this.transactionDefinition != null) {
+			poller.setPropagationBehavior(this.transactionDefinition.getPropagationBehavior());
+			poller.setIsolationLevel(this.transactionDefinition.getIsolationLevel());
+			poller.setTransactionReadOnly(this.transactionDefinition.isReadOnly());
+			poller.setTransactionTimeout(this.transactionDefinition.getTimeout());
+		}
+	}
+
 	public String toString() {
 		return (this.name != null) ? this.name : super.toString();
 	}

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

@@ -34,7 +34,7 @@ import org.springframework.integration.message.selector.MessageSelector;
 /**
  * @author Mark Fisher
  */
-public abstract class AbstractInOutEndpoint extends AbstractEndpoint {
+public abstract class AbstractInOutEndpoint extends AbstractMessageConsumingEndpoint {
 
 	private MessageChannel outputChannel;
 
@@ -73,11 +73,11 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint {
 	}
 
 	@Override
-	protected boolean sendInternal(Message<?> message) {
+	protected void processMessage(Message<?> message) {
 		for (EndpointInterceptor interceptor : this.interceptors) {
 			message = interceptor.preHandle(message);
 			if (message == null) {
-				return false;
+				return;
 			}
 		}
 		if (!this.supports(message)) {
@@ -89,7 +89,7 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint {
 				throw new MessageHandlingException(message, "endpoint '" + this.getName()
 						+ " requires a reply, but no reply was received");
 			}
-			return true;
+			return;
 		}
 		Message<?> reply = null;
 		if (result instanceof Message && result.equals(message)) {
@@ -106,10 +106,9 @@ public abstract class AbstractInOutEndpoint extends AbstractEndpoint {
 				boolean sent = this.sendReplyMessage(nextReply, replyChannel);
 				sentAtLeastOne = (sentAtLeastOne || sent);
 			}
-			return sentAtLeastOne;
 		}
 		else {
-			return this.sendReplyMessage(reply, replyChannel);
+			this.sendReplyMessage(reply, replyChannel);
 		}
 	}
 

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

@@ -0,0 +1,147 @@
+/*
+ * 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.context.Lifecycle;
+import org.springframework.integration.ConfigurationException;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.PollableChannel;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageConsumer;
+import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.integration.message.MessagingException;
+import org.springframework.integration.message.SubscribableSource;
+import org.springframework.integration.scheduling.PollingSchedule;
+import org.springframework.integration.scheduling.Schedule;
+
+/**
+ * The base class for Message Endpoint implementations that consume Messages.
+ * 
+ * @author Mark Fisher
+ */
+public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint implements MessageConsumer, Lifecycle {
+
+	private volatile MessageChannel inputChannel;
+
+	private volatile Schedule schedule = new PollingSchedule(0);
+
+	private volatile ChannelPoller poller;
+
+	private volatile int maxMessagesPerPoll = -1;
+
+	private volatile boolean initialized;
+
+	private volatile boolean running;
+
+	private final Object lifecycleMonitor = new Object();
+
+
+	public void setInputChannel(MessageChannel inputChannel) {
+		this.inputChannel = inputChannel;
+	}
+
+	public void setSchedule(Schedule schedule) {
+		this.schedule = schedule;
+	}
+
+	public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
+		this.maxMessagesPerPoll = maxMessagesPerPoll;
+		if (this.poller != null) {
+			this.poller.setMaxMessagesPerPoll(maxMessagesPerPoll);
+		}
+	}
+
+	public final boolean isRunning() {
+		return this.running;
+	}
+
+	@Override
+	protected void initialize()  throws Exception {
+		synchronized (this.lifecycleMonitor) {
+			if (this.inputChannel instanceof PollableChannel && this.poller == null) {
+				this.poller = new ChannelPoller((PollableChannel) this.inputChannel, this.schedule);
+				this.poller.setMaxMessagesPerPoll(this.maxMessagesPerPoll);
+				this.configureTransactionSettingsForPoller(this.poller);
+				this.poller.subscribe(this);
+			}
+			this.initialized = true;
+		}
+	}
+
+	public final void start() {
+		synchronized (this.lifecycleMonitor) {
+			if (this.running) {
+				return;
+			}
+			if (!this.initialized) {
+				this.afterPropertiesSet();
+			}
+			if (this.inputChannel == null) {
+				throw new ConfigurationException("failed to start endpoint, inputChannel is required");
+			}
+			if (this.inputChannel instanceof SubscribableSource) {
+				((SubscribableSource) inputChannel).subscribe(this);
+			}
+			else if (this.inputChannel instanceof PollableChannel) {
+				if (this.getTaskScheduler() == null) {
+					throw new ConfigurationException("failed to start endpoint, no taskScheduler available");
+				}
+				this.getTaskScheduler().schedule(poller);
+			}
+			this.running = true;
+		}
+	}
+
+	public final void stop() {
+		synchronized (this.lifecycleMonitor) {
+			if (!this.running) {
+				return;
+			}
+			if (this.inputChannel instanceof SubscribableSource) {
+				((SubscribableSource) inputChannel).unsubscribe(this);
+			}
+			else if (this.poller != null) {
+				this.getTaskScheduler().cancel(poller, true);
+			}
+			this.running = false;
+		}
+	}
+
+	public final void onMessage(Message<?> message) {
+		if (message == null || message.getPayload() == null) {
+			throw new IllegalArgumentException("Message and its payload must not be null");
+		}
+		if (this.logger.isDebugEnabled()) {
+			this.logger.debug("endpoint '" + this + "' processing message: " + message);
+		}
+		try {
+			this.processMessage(message);
+		}
+		catch (Exception e) {
+			if (e instanceof MessagingException) {
+				this.handleException((MessagingException) e);
+			}
+			else {
+				this.handleException(new MessageHandlingException(message,
+						"failure occurred in endpoint '" + this.toString() + "'", e));
+			}
+		}
+	}
+
+	protected abstract void processMessage(Message<?> message);
+
+}

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

@@ -18,11 +18,11 @@ package org.springframework.integration.endpoint;
 
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.core.task.TaskExecutor;
-import org.springframework.integration.message.SubscribableSource;
 import org.springframework.integration.scheduling.SchedulableTask;
 import org.springframework.integration.scheduling.Schedule;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.DefaultTransactionDefinition;
 import org.springframework.transaction.support.TransactionCallback;
 import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.util.Assert;
@@ -30,7 +30,7 @@ import org.springframework.util.Assert;
 /**
  * @author Mark Fisher
  */
-public abstract class AbstractPoller implements SubscribableSource, SchedulableTask, InitializingBean {
+public abstract class AbstractPoller implements SchedulableTask, InitializingBean {
 
 	public static final int MAX_MESSAGES_UNBOUNDED = -1;
 
@@ -45,11 +45,11 @@ public abstract class AbstractPoller implements SubscribableSource, SchedulableT
 
 	private volatile TransactionTemplate transactionTemplate;
 
-	private volatile String propagationBehaviorName = "PROPAGATION_REQUIRED";
+	private volatile int propagationBehavior = DefaultTransactionDefinition.PROPAGATION_REQUIRED;
 
-	private volatile String isolationLevelName = "ISOLATION_DEFAULT";
+	private volatile int isolationLevel = DefaultTransactionDefinition.ISOLATION_DEFAULT;
 
-	private volatile int transactionTimeout = -1;
+	private volatile int transactionTimeout = DefaultTransactionDefinition.TIMEOUT_DEFAULT;
 
 	private volatile boolean readOnly = false;
 
@@ -94,12 +94,12 @@ public abstract class AbstractPoller implements SubscribableSource, SchedulableT
 		this.transactionManager = transactionManager;
 	}
 
-	public void setPropagationBehaviorName(String propagationBehaviorName) {
-		this.propagationBehaviorName = propagationBehaviorName;
+	public void setPropagationBehavior(int propagationBehavior) {
+		this.propagationBehavior = propagationBehavior;
 	}
 
-	public void setIsolationLevelName(String isolationLevelName) {
-		this.isolationLevelName = isolationLevelName;
+	public void setIsolationLevel(int isolationLevel) {
+		this.isolationLevel = isolationLevel;
 	}
 
 	public void setTransactionTimeout(int transactionTimeout) {
@@ -124,8 +124,8 @@ public abstract class AbstractPoller implements SubscribableSource, SchedulableT
 			}
 			if (this.transactionManager != null) {
 				TransactionTemplate template = new TransactionTemplate(this.transactionManager);
-				template.setPropagationBehaviorName(this.propagationBehaviorName);
-				template.setIsolationLevelName(this.isolationLevelName);
+				template.setPropagationBehavior(this.propagationBehavior);
+				template.setIsolationLevel(this.isolationLevel);
 				template.setTimeout(this.transactionTimeout);
 				template.setReadOnly(this.readOnly);
 				this.transactionTemplate = template;

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

@@ -19,6 +19,7 @@ package org.springframework.integration.endpoint;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.dispatcher.SimpleDispatcher;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.message.SubscribableSource;
 import org.springframework.integration.scheduling.Schedule;
 import org.springframework.util.Assert;
@@ -50,12 +51,12 @@ public class ChannelPoller extends AbstractPoller implements SubscribableSource
 		this.receiveTimeout = receiveTimeout;
 	}
 
-	public boolean subscribe(MessageEndpoint endpoint) {
-		return this.dispatcher.subscribe(endpoint);
+	public boolean subscribe(MessageConsumer consumer) {
+		return this.dispatcher.subscribe(consumer);
 	}
 
-	public boolean unsubscribe(MessageEndpoint endpoint) {
-		return this.dispatcher.unsubscribe(endpoint);
+	public boolean unsubscribe(MessageConsumer consumer) {
+		return this.dispatcher.unsubscribe(consumer);
 	}
 
 	@Override

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

@@ -16,11 +16,12 @@
 
 package org.springframework.integration.endpoint;
 
+import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageDeliveryAware;
-import org.springframework.integration.message.MessageDeliveryException;
-import org.springframework.integration.message.MessagingException;
+import org.springframework.integration.message.MethodInvokingSource;
+import org.springframework.integration.message.PollableSource;
+import org.springframework.integration.scheduling.Schedule;
+import org.springframework.integration.scheduling.TaskScheduler;
 
 /**
  * A Channel Adapter implementation for connecting a
@@ -29,33 +30,75 @@ import org.springframework.integration.message.MessagingException;
  * 
  * @author Mark Fisher
  */
-public class InboundChannelAdapter extends AbstractEndpoint {
+public class InboundChannelAdapter extends AbstractEndpoint implements Lifecycle {
 
-	private MessageChannel channel;
+	private volatile PollableSource<?> source;
 
+	private volatile MessageChannel channel;
+
+	private volatile Schedule schedule;
+
+	private volatile SourcePoller poller;
+
+	private volatile int maxMessagesPerPoll = -1;
+
+	private volatile boolean running;
+
+	private final Object lifecycleMonitor = new Object();
+
+
+	public void setSource(PollableSource<?> source) {
+		this.source = source;
+	}
 
 	public void setChannel(MessageChannel channel) {
 		this.channel = channel;
 	}
 
-	@Override
-	protected boolean sendInternal(Message<?> message) {
-		if (this.channel == null) {
-			throw new MessageDeliveryException(message, "no channel has been provided");
+	public void setSchedule(Schedule schedule) {
+		this.schedule = schedule;
+	}
+
+	public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
+		this.maxMessagesPerPoll = maxMessagesPerPoll;
+		if (this.poller != null) {
+			this.poller.setMaxMessagesPerPoll(maxMessagesPerPoll);
 		}
-		try {
-			boolean sent = this.getMessageExchangeTemplate().send(message, this.channel);
-			if (sent && this.getSource() instanceof MessageDeliveryAware) {
-				((MessageDeliveryAware) this.getSource()).onSend(message);
+	}
+
+	public final boolean isRunning() {
+		return this.running;
+	}
+
+	public final void start() {
+		synchronized (this.lifecycleMonitor) {
+			if (this.running) {
+				return;
+			}
+			this.poller = new SourcePoller(source, channel, schedule);
+			if (maxMessagesPerPoll < 0 && source instanceof MethodInvokingSource) {
+				// the default is 1 since a MethodInvokingSource might return a non-null value
+				// every time it is invoked, thus producing an infinite number of messages per poll
+				maxMessagesPerPoll = 1;
+			}
+			this.configureTransactionSettingsForPoller(this.poller);
+			this.poller.setMaxMessagesPerPoll(maxMessagesPerPoll);
+			TaskScheduler taskScheduler = this.getTaskScheduler();
+			if (taskScheduler != null) {
+				taskScheduler.schedule(this.poller);
 			}
-			return sent;
 		}
-		catch (Exception e) {
-			if (this.getSource() instanceof MessageDeliveryAware) {
-				((MessageDeliveryAware) this.getSource()).onFailure(message, e);
+	}
+
+	public final void stop() {
+		synchronized (this.lifecycleMonitor) {
+			if (!this.running) {
+				return;
+			}
+			TaskScheduler taskScheduler = this.getTaskScheduler();
+			if (taskScheduler != null) {
+				taskScheduler.cancel(this.poller, true);
 			}
-			throw (e instanceof MessagingException) ? (MessagingException) e
-					: new MessageDeliveryException(message, "channel adapter failed to send message to target", e);
 		}
 	}
 

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

@@ -16,8 +16,6 @@
 
 package org.springframework.integration.endpoint;
 
-import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageSource;
 
 /**
  * Base interface for message endpoints.
@@ -28,8 +26,4 @@ public interface MessageEndpoint {
 
 	String getName();
 
-	MessageSource<?> getSource();
-
-	boolean send(Message<?> message);
-
 }

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

@@ -18,6 +18,7 @@ package org.springframework.integration.endpoint;
 
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageDeliveryException;
 import org.springframework.integration.message.MessageTarget;
 import org.springframework.util.Assert;
 
@@ -27,7 +28,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class OutboundChannelAdapter extends AbstractEndpoint {
+public class OutboundChannelAdapter extends AbstractMessageConsumingEndpoint {
 
 	private final MessageTarget target;
 
@@ -39,8 +40,10 @@ public class OutboundChannelAdapter extends AbstractEndpoint {
 
 
 	@Override
-	protected boolean sendInternal(Message<?> message) {
-		return this.target.send(message);
+	protected void processMessage(Message<?> message) {
+		if (!this.target.send(message)) {
+			throw new MessageDeliveryException(message, "failed to deliver Message to target");
+		}
 	}
 
 }

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

@@ -50,6 +50,7 @@ public class ServiceActivatorEndpoint extends AbstractInOutEndpoint {
 
 	@Override
 	protected void initialize() throws Exception {
+		super.initialize();
 		if (this.invoker instanceof InitializingBean) {
 			((InitializingBean) this.invoker).afterPropertiesSet();
 		}

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

@@ -16,30 +16,34 @@
 
 package org.springframework.integration.endpoint;
 
-import org.springframework.integration.dispatcher.SimpleDispatcher;
+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.MessagingException;
 import org.springframework.integration.message.PollableSource;
-import org.springframework.integration.message.SubscribableSource;
 import org.springframework.integration.scheduling.Schedule;
 import org.springframework.util.Assert;
 
 /**
  * @author Mark Fisher
  */
-public class SourcePoller extends AbstractPoller implements SubscribableSource {
+public class SourcePoller extends AbstractPoller {
 
 	private final PollableSource<?> source;
 
-	private final SimpleDispatcher dispatcher = new SimpleDispatcher();
+	private final MessageChannel channel;
 
 	private volatile long receiveTimeout = 1000;
 
 
-	public SourcePoller(PollableSource<?> source, Schedule schedule) {
+	public SourcePoller(PollableSource<?> source, MessageChannel channel, Schedule schedule) {
 		super(schedule);
 		Assert.notNull(source, "source must not be null");
+		Assert.notNull(channel, "channel must not be null");
 		this.source = source;
+		this.channel = channel;
 	}
 
 
@@ -54,14 +58,6 @@ public class SourcePoller extends AbstractPoller implements SubscribableSource {
 		this.receiveTimeout = receiveTimeout;
 	}
 
-	public boolean subscribe(MessageEndpoint endpoint) {
-		return this.dispatcher.subscribe(endpoint);
-	}
-
-	public boolean unsubscribe(MessageEndpoint endpoint) {
-		return this.dispatcher.unsubscribe(endpoint);
-	}
-
 	@Override
 	protected boolean doPoll() {
 		Message<?> message = (this.receiveTimeout >= 0 && this.source instanceof BlockingSource)
@@ -70,7 +66,20 @@ public class SourcePoller extends AbstractPoller implements SubscribableSource {
 		if (message == null) {
 			return false;
 		}
-		return this.dispatcher.dispatch(message);
+		try {
+			boolean sent = this.channel.send(message); 
+			if (sent && this.source instanceof MessageDeliveryAware) {
+				((MessageDeliveryAware) this.source).onSend(message);
+			}
+			return sent;
+		}
+		catch (Exception e) {
+			if (this.source instanceof MessageDeliveryAware) {
+				((MessageDeliveryAware) this.source).onFailure(message, e);
+			}
+			throw (e instanceof MessagingException) ? (MessagingException) e
+					: new MessageDeliveryException(message, "source poller failed to send message to channel", e);
+		}
 	}
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java

@@ -205,7 +205,7 @@ public class SimpleMessagingGateway extends MessagingGatewaySupport implements M
 			}
 			ReplyMessageCorrelator correlator = new ReplyMessageCorrelator(this.replyMapCapacity);
 			correlator.setBeanName("internal.correlator." + this);
-			correlator.setSource(this.replyChannel);
+			correlator.setInputChannel(this.replyChannel);
 			correlator.afterPropertiesSet();
 			this.endpointRegistry.registerEndpoint(correlator);
 			this.replyMessageCorrelator = correlator;

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

@@ -16,8 +16,6 @@
 
 package org.springframework.integration.message;
 
-import org.springframework.integration.endpoint.MessageEndpoint;
-
 /**
  * Interface for any source of messages that accepts subscribers.
  * 
@@ -26,13 +24,13 @@ import org.springframework.integration.endpoint.MessageEndpoint;
 public interface SubscribableSource extends MessageSource {
 
 	/**
-	 * Register a {@link MessageEndpoint} as a subscriber to this source.
+	 * Register a {@link MessageConsumer} as a subscriber to this source.
 	 */
-	boolean subscribe(MessageEndpoint endpoint);
+	boolean subscribe(MessageConsumer consumer);
 
 	/**
-	 * Remove a {@link MessageEndpoint} from the subscribers of this source.
+	 * Remove a {@link MessageConsumer} from the subscribers of this source.
 	 */
-	boolean unsubscribe(MessageEndpoint endpoint);
+	boolean unsubscribe(MessageConsumer consumer);
 
 }

org.springframework.integration/src/main/java/org/springframework/integration/router/RouterEndpoint.java

@@ -21,7 +21,7 @@ import java.util.Collection;
 import org.springframework.integration.channel.ChannelRegistry;
 import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.endpoint.AbstractMessageConsumingEndpoint;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageDeliveryException;
 import org.springframework.integration.message.MessageExchangeTemplate;
@@ -30,7 +30,7 @@ import org.springframework.util.Assert;
 /**
  * @author Mark Fisher
  */
-public class RouterEndpoint extends AbstractEndpoint {
+public class RouterEndpoint extends AbstractMessageConsumingEndpoint {
 
 	private final ChannelResolver channelResolver;
 
@@ -78,7 +78,7 @@ public class RouterEndpoint extends AbstractEndpoint {
 	}
 
 	@Override
-	protected boolean sendInternal(Message<?> message) {
+	protected void processMessage(Message<?> message) {
 		boolean sent = false;
 		Collection<MessageChannel> results = this.channelResolver.resolveChannels(message);
 		if (results != null) {
@@ -99,7 +99,6 @@ public class RouterEndpoint extends AbstractEndpoint {
 						"no target resolved by router and no default output channel defined");
 			}
 		}
-		return sent;
 	}
 
 }