Refactored existing Message-consuming endpoints to only implement MessageConsumer (not MessageEndpoint). Now, either a PollingConsumerEndpoint or SubscribingConsumerEndpoint delegates to the MessageConsumer thereby separating the Lifecycle responsibilities and configuration settings (trigger, transactions, etc) since they are different for polling vs. subscribing and not relevant for simply consuming Messages. Essentially all MessageConsumers are now "event-driven" since a "polling consumer" is actually handled by the PollingConsumerEndpoint class. The next refactoring step involves renaming several components to clarify this endpoint vs. consumer distinction.

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

@@ -27,26 +27,29 @@ import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.channel.BlockingChannel;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.endpoint.AbstractMessageHandlingEndpoint;
-import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.message.MessageHandlingException;
 import org.springframework.integration.scheduling.IntervalTrigger;
+import org.springframework.integration.scheduling.TaskScheduler;
 import org.springframework.integration.scheduling.TaskSchedulerAware;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
 
 /**
- * Base class for {@link MessageBarrier}-based MessageHandlers.
- * A {@link MessageEndpoint} implementation that waits for a group of
+ * Base class for {@link MessageBarrier}-based Message Consumers.
+ * A {@link MessageConsumer} implementation that waits for a group of
  * {@link Message Messages} to arrive and processes them together.
- * Uses a {@link MessageBarr  ier} to store messages and to decide how
+ * Uses a {@link MessageBarrier} to store messages and to decide how
  * the messages should be released.
  * <p>
- * Each {@link Message} that is received by this endpoint will be associated with
+ * Each {@link Message} that is received by this consumer will be associated with
  * a group based upon the '<code>correlationId</code>' property of its
  * header. If no such property is available, a {@link MessageHandlingException}
  * will be thrown.
@@ -60,7 +63,7 @@ import org.springframework.util.ObjectUtils;
  * @author Mark Fisher
  * @author Marius Bogoevici
  */
-public abstract class AbstractMessageBarrierEndpoint extends AbstractMessageHandlingEndpoint implements TaskSchedulerAware {
+public abstract class AbstractMessageBarrierEndpoint extends AbstractMessageHandlingEndpoint implements TaskSchedulerAware, InitializingBean {
 
 	public final static long DEFAULT_SEND_TIMEOUT = 1000;
 
@@ -91,8 +94,11 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractMessageHand
 
 	private volatile boolean initialized;
 
+	private TaskScheduler taskScheduler;
+
 	private ScheduledFuture<?> reaperFutureTask;
 
+
 	/**
 	 * Specify a channel for sending Messages that arrive after their aggregation
 	 * group has either completed or timed-out.
@@ -141,26 +147,32 @@ public abstract class AbstractMessageBarrierEndpoint extends AbstractMessageHand
 		this.timeout = timeout;
 	}
 
-	/**
-	 * Initialize this endpoint.
-	 */
-	@Override
-	protected void initialize()  throws Exception {
-		super.initialize();
+	public void setTaskScheduler(TaskScheduler taskScheduler) {
+		this.taskScheduler = taskScheduler;
+	}
+
+	public void afterPropertiesSet() {
 		this.trackedCorrelationIds = new ArrayBlockingQueue<Object>(this.trackedCorrelationIdCapacity);
 		this.initialized = true;
 	}
-	
-	@Override
-	protected void onStart() {
-		super.onStart();
-		this.reaperFutureTask = this.getTaskScheduler().schedule(new ReaperTask(), new IntervalTrigger(reaperInterval, TimeUnit.MILLISECONDS));
+
+	public boolean isRunning() {
+		return this.reaperFutureTask != null;
 	}
 
-	@Override
-	protected void onStop() {
-		super.onStop();
-		this.reaperFutureTask.cancel(true);
+	public void start() {
+		if (this.isRunning()) {
+			return;
+		}
+		Assert.state(this.taskScheduler != null, "TaskScheduler must not be null");
+		this.reaperFutureTask = this.taskScheduler.schedule(
+				new ReaperTask(), new IntervalTrigger(this.reaperInterval, TimeUnit.MILLISECONDS));
+	}
+
+	public void stop() {
+		if (this.isRunning()) {
+			this.reaperFutureTask.cancel(true);
+		}
 	}
 
 	@Override

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

@@ -18,13 +18,11 @@ package org.springframework.integration.config;
 
 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.AbstractSingleBeanDefinitionParser;
 import org.springframework.beans.factory.xml.ParserContext;
-import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.integration.endpoint.config.ConsumerEndpointFactoryBean;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 import org.springframework.util.xml.DomUtils;
@@ -40,8 +38,6 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio
 
 	protected static final String METHOD_ATTRIBUTE = "method";
 
-	protected static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel";
-
 	protected static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
 
 	private static final String POLLER_ELEMENT = "poller";
@@ -51,20 +47,9 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio
 	private static final String ERROR_HANDLER_ATTRIBUTE = "error-handler";
 
 
-	/**
-	 * Subclasses may override this method to determine whether the endpoint
-	 * type should create an adapter. If so, the "ref" attribute will be
-	 * required, and the "method" attribute will typically be used as well.
-	 * 
-	 * <p>The default is <em>true</em>.
-	 */
-	protected boolean shouldCreateAdapter(Element element) {
-		return true;
-	}
-
 	@Override
 	protected final Class<?> getBeanClass(Element element) {
-		return this.getEndpointClass();
+		return ConsumerEndpointFactoryBean.class;
 	}
 
 	@Override
@@ -77,22 +62,42 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio
 		return true;
 	}
 
-	@Override
-	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
-		if (this.shouldCreateAdapter(element)) {
-			String ref = element.getAttribute(REF_ATTRIBUTE);
-			Assert.hasText(ref, "The '" + REF_ATTRIBUTE + "' attribute is required.");
-			if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
-				String method = element.getAttribute(METHOD_ATTRIBUTE);
-				String adapterBeanName = this.parseAdapter(ref, method, element, parserContext);
-				builder.addConstructorArgReference(adapterBeanName);
-			}
-			else {
-				builder.addConstructorArgReference(ref);
-			}
+	/**
+	 * Parse the MessageConsumer.
+	 */
+	protected abstract BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext);
+
+	protected String getInputChannelAttributeName() {
+		return "input-channel";
+	}
+
+	protected String parseAdapter(Element element, ParserContext parserContext, Class<?> adapterClass) {
+		String ref = element.getAttribute(REF_ATTRIBUTE);
+		Assert.hasText(ref, "The '" + REF_ATTRIBUTE + "' attribute is required.");
+		if (StringUtils.hasText(element.getAttribute(METHOD_ATTRIBUTE))) {
+			BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(adapterClass);
+			String method = element.getAttribute(METHOD_ATTRIBUTE);
+			builder.addConstructorArgReference(ref);
+			builder.addConstructorArgValue(method);
+			return BeanDefinitionReaderUtils.registerWithGeneratedName(
+					builder.getBeanDefinition(), parserContext.getRegistry());
 		}
-		String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE);
-		Assert.hasText(inputChannel, "the '" + INPUT_CHANNEL_ATTRIBUTE + "' attribute is required");
+		return ref;
+	}
+
+	@Override
+	protected final void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+		BeanDefinitionBuilder consumerBuilder = this.parseConsumer(element, parserContext);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(consumerBuilder, element, OUTPUT_CHANNEL_ATTRIBUTE);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(consumerBuilder, element, SELECTOR_ATTRIBUTE);
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(consumerBuilder, element, ERROR_HANDLER_ATTRIBUTE);
+		String consumerBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
+				consumerBuilder.getBeanDefinition(), parserContext.getRegistry());
+		builder.addConstructorArgReference(consumerBeanName);
+		String inputChannelAttributeName = this.getInputChannelAttributeName();
+		String inputChannelName = element.getAttribute(inputChannelAttributeName);
+		Assert.hasText(inputChannelName, "the '" + inputChannelAttributeName + "' attribute is required");
+		builder.addPropertyValue("inputChannelName", inputChannelName);
 		Element pollerElement = DomUtils.getChildElementByTagName(element, POLLER_ELEMENT);
 		if (pollerElement != null) {
 			IntegrationNamespaceUtils.configureTrigger(pollerElement, builder);
@@ -102,41 +107,13 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio
 			}
 			IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, pollerElement, "task-executor");
 		}
-		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);
 		this.postProcess(element, parserContext, builder);
 	}
 
-	private String parseAdapter(String ref, String method, Element element, ParserContext parserContext) {
-		Class<?> adapterClass = this.getMethodInvokingAdapterClass();
-		Assert.state(adapterClass != null,
-				"Parser needs to create an adapter but 'getMethodInvokingAdapterClass()' returned null.");
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(adapterClass);
-		builder.addConstructorArgReference(ref);
-		builder.addConstructorArgValue(method);
-		String adapterBeanName = BeanDefinitionReaderUtils.generateBeanName(
-				builder.getBeanDefinition(), parserContext.getRegistry());
-		BeanDefinitionHolder holder = new BeanDefinitionHolder(builder.getBeanDefinition(), adapterBeanName);
-		parserContext.registerBeanComponent(new BeanComponentDefinition(holder));
-		return adapterBeanName;
-	}
-
 	/**
 	 * Subclasses may implement this method to provide additional configuration.
 	 */
 	protected void postProcess(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
 	}
 
-	/**
-	 * Subclasses must override this if the adapted object is created from
-	 * the "ref" and "method" attribute values.
-	 */
-	protected Class<?> getMethodInvokingAdapterClass() {
-		return null;
-	}
-
-	protected abstract Class<? extends MessageEndpoint> getEndpointClass();
-
 }

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

@@ -22,7 +22,7 @@ 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.xml.ParserContext;
-import org.springframework.integration.endpoint.OutboundChannelAdapter;
+import org.springframework.integration.endpoint.config.ConsumerEndpointFactoryBean;
 import org.springframework.util.Assert;
 import org.springframework.util.xml.DomUtils;
 
@@ -36,20 +36,18 @@ public abstract class AbstractOutboundChannelAdapterParser extends AbstractChann
 	@Override
 	protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
 		Element pollerElement = DomUtils.getChildElementByTagName(element, "poller");
-		BeanDefinitionBuilder adapterBuilder = null;
-		adapterBuilder =  BeanDefinitionBuilder.genericBeanDefinition(OutboundChannelAdapter.class);
-		String consumerBeanName = this.parseAndRegisterConsumer(element, parserContext);
-		adapterBuilder.addConstructorArgReference(consumerBeanName);
+		BeanDefinitionBuilder builder =  BeanDefinitionBuilder.genericBeanDefinition(ConsumerEndpointFactoryBean.class);
+		builder.addConstructorArgReference(this.parseAndRegisterConsumer(element, parserContext));
 		if (pollerElement != null) {
 			Assert.hasText(channelName, "outbound channel adapter with a 'poller' requires a 'channel' to poll");
-			IntegrationNamespaceUtils.configureTrigger(pollerElement, adapterBuilder);
+			IntegrationNamespaceUtils.configureTrigger(pollerElement, builder);
 			Element txElement = DomUtils.getChildElementByTagName(pollerElement, "transactional");
 			if (txElement != null) {
-				IntegrationNamespaceUtils.configureTransactionAttributes(txElement, adapterBuilder);
+				IntegrationNamespaceUtils.configureTransactionAttributes(txElement, builder);
 			}
 		}
-		adapterBuilder.addPropertyReference("inputChannel", channelName);
-		return adapterBuilder.getBeanDefinition();
+		builder.addPropertyValue("inputChannelName", channelName);
+		return builder.getBeanDefinition();
 	}
 
 	/**

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

@@ -22,9 +22,8 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.aggregator.AggregatorEndpoint;
-import org.springframework.integration.aggregator.MethodInvokingAggregator;
 import org.springframework.integration.aggregator.CompletionStrategyAdapter;
-import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.integration.aggregator.MethodInvokingAggregator;
 import org.springframework.util.StringUtils;
 
 /**
@@ -36,40 +35,35 @@ import org.springframework.util.StringUtils;
  */
 public class AggregatorParser extends AbstractEndpointParser {
 
-	public static final String COMPLETION_STRATEGY_REF_ATTRIBUTE = "completion-strategy";
+	private static final String COMPLETION_STRATEGY_REF_ATTRIBUTE = "completion-strategy";
 
-	public static final String COMPLETION_STRATEGY_METHOD_ATTRIBUTE = "completion-strategy-method";
+	private static final String COMPLETION_STRATEGY_METHOD_ATTRIBUTE = "completion-strategy-method";
 
-	public static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";
+	private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";
 
-	public static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout";
+	private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout";
 
-	public static final String SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE = "send-partial-result-on-timeout";
+	private static final String SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE = "send-partial-result-on-timeout";
 
-	public static final String REAPER_INTERVAL_ATTRIBUTE = "reaper-interval";
+	private static final String REAPER_INTERVAL_ATTRIBUTE = "reaper-interval";
 
-	public static final String TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE = "tracked-correlation-id-capacity";
+	private static final String TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE = "tracked-correlation-id-capacity";
 
-	public static final String TIMEOUT_ATTRIBUTE = "timeout";
+	private static final String TIMEOUT_ATTRIBUTE = "timeout";
 
 	private static final String COMPLETION_STRATEGY_PROPERTY = "completionStrategy";
 
-	public static final String AGGREGATOR_ELEMENT = "aggregator";
-
 
 	@Override
-	protected Class<? extends MessageEndpoint> getEndpointClass() {
-		return AggregatorEndpoint.class;
-	}
-
-	@Override
-	protected Class<?> getMethodInvokingAdapterClass() {
-		return MethodInvokingAggregator.class;
-	}
-
-	@Override
-	protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
-		super.doParse(element, parserContext, builder);
+	protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(AggregatorEndpoint.class);
+		builder.addConstructorArgReference(this.parseAdapter(element, parserContext, MethodInvokingAggregator.class));
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, REAPER_INTERVAL_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE);
 		final String completionStrategyRef = element.getAttribute(COMPLETION_STRATEGY_REF_ATTRIBUTE);
 		final String completionStrategyMethod = element.getAttribute(COMPLETION_STRATEGY_METHOD_ATTRIBUTE);
 		if (StringUtils.hasText(completionStrategyRef)) {
@@ -82,12 +76,7 @@ public class AggregatorParser extends AbstractEndpointParser {
 				builder.addPropertyReference(COMPLETION_STRATEGY_PROPERTY, completionStrategyRef);
 			}
 		}
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, REAPER_INTERVAL_ATTRIBUTE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE);
+		return builder;
 	}
 
 	private String createCompletionStrategyAdapter(String ref, String method, ParserContext parserContext) {

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

@@ -16,7 +16,10 @@
 
 package org.springframework.integration.config;
 
-import org.springframework.integration.endpoint.MessageEndpoint;
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.filter.FilterEndpoint;
 import org.springframework.integration.filter.MethodInvokingSelector;
 
@@ -28,13 +31,10 @@ import org.springframework.integration.filter.MethodInvokingSelector;
 public class FilterParser extends AbstractEndpointParser {
 
 	@Override
-	protected Class<? extends MessageEndpoint> getEndpointClass() {
-		return FilterEndpoint.class;
-	}
-
-	@Override
-	protected Class<?> getMethodInvokingAdapterClass() {
-		return MethodInvokingSelector.class;
+	protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(FilterEndpoint.class);
+		builder.addConstructorArgReference(this.parseAdapter(element, parserContext, MethodInvokingSelector.class));
+		return builder;
 	}
 
 }

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

@@ -19,7 +19,7 @@ package org.springframework.integration.config;
 import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.aggregator.ResequencerEndpoint;
 
 /**
@@ -27,33 +27,34 @@ import org.springframework.integration.aggregator.ResequencerEndpoint;
  * 
  * @author Marius Bogoevici
  */
-public class ResequencerParser extends AbstractSimpleBeanDefinitionParser {
+public class ResequencerParser extends AbstractEndpointParser {
 
-	public static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel";
+	private static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";
 
-	public static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
+	private static final String SEND_TIMEOUT_ATTRIBUTE = "send-timeout";
 
-	public static final String DISCARD_CHANNEL_ATTRIBUTE = "discard-channel";
+	private static final String RELEASE_PARTIAL_SEQUENCES = "release-partial-sequences";
+
+	private static final String SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE = "send-partial-result-on-timeout";
+
+	private static final String REAPER_INTERVAL_ATTRIBUTE = "reaper-interval";
+
+	private static final String TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE = "tracked-correlation-id-capacity";
+
+	private static final String TIMEOUT_ATTRIBUTE = "timeout";
 
 
 	@Override
-	protected Class<?> getBeanClass(Element element) {
-		return ResequencerEndpoint.class;
-	}
-
-	@Override
-	protected boolean isEligibleAttribute(String attributeName) {
-		return !INPUT_CHANNEL_ATTRIBUTE.equals(attributeName)
-				&&!OUTPUT_CHANNEL_ATTRIBUTE.equals(attributeName)
-				&& !DISCARD_CHANNEL_ATTRIBUTE.equals(attributeName)
-				&& super.isEligibleAttribute(attributeName);
-	}
-
-	@Override
-	protected void postProcess(BeanDefinitionBuilder builder, Element element) {
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, INPUT_CHANNEL_ATTRIBUTE);
-		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, OUTPUT_CHANNEL_ATTRIBUTE);
+	protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ResequencerEndpoint.class);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, DISCARD_CHANNEL_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_TIMEOUT_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, RELEASE_PARTIAL_SEQUENCES);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, SEND_PARTIAL_RESULT_ON_TIMEOUT_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, REAPER_INTERVAL_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TRACKED_CORRELATION_ID_CAPACITY_ATTRIBUTE);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, TIMEOUT_ATTRIBUTE);
+		return builder;
 	}
 
 }

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

@@ -20,7 +20,6 @@ import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
-import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.router.MethodInvokingChannelResolver;
 import org.springframework.integration.router.RouterEndpoint;
 
@@ -32,18 +31,13 @@ import org.springframework.integration.router.RouterEndpoint;
 public class RouterParser extends AbstractEndpointParser {
 
 	@Override
-	protected Class<? extends MessageEndpoint> getEndpointClass() {
-		return RouterEndpoint.class;
-	}
-
-	@Override
-	protected Class<?> getMethodInvokingAdapterClass() {
-		return MethodInvokingChannelResolver.class;
-	}
-
-	@Override
-	protected void postProcess(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+	protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
+		String adapterBeanName = this.parseAdapter(element, parserContext, MethodInvokingChannelResolver.class);
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RouterEndpoint.class);
+		builder.addConstructorArgReference(adapterBeanName);
+		builder.addPropertyReference("channelRegistry", MessageBusParser.MESSAGE_BUS_BEAN_NAME);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "default-output-channel");
+		return builder;
 	}
 
 }

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

@@ -16,7 +16,10 @@
 
 package org.springframework.integration.config;
 
-import org.springframework.integration.endpoint.MessageEndpoint;
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
 import org.springframework.integration.message.MessageMappingMethodInvoker;
 
@@ -28,13 +31,11 @@ import org.springframework.integration.message.MessageMappingMethodInvoker;
 public class ServiceActivatorParser extends AbstractEndpointParser {
 
 	@Override
-	protected Class<? extends MessageEndpoint> getEndpointClass() {
-		return ServiceActivatorEndpoint.class;
-	}
-
-	@Override
-	protected Class<?> getMethodInvokingAdapterClass() {
-		return MessageMappingMethodInvoker.class;
+	protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ServiceActivatorEndpoint.class);
+		String constructorArg = this.parseAdapter(element, parserContext, MessageMappingMethodInvoker.class);
+		builder.addConstructorArgReference(constructorArg);
+		return builder;
 	}
 
 }

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

@@ -18,7 +18,8 @@ package org.springframework.integration.config;
 
 import org.w3c.dom.Element;
 
-import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.splitter.MethodInvokingSplitter;
 import org.springframework.integration.splitter.SplitterEndpoint;
 
@@ -30,18 +31,13 @@ import org.springframework.integration.splitter.SplitterEndpoint;
 public class SplitterParser extends AbstractEndpointParser {
 
 	@Override
-	protected boolean shouldCreateAdapter(Element element) {
-		return element.hasAttribute("ref");
-	}
-
-	@Override
-	protected Class<? extends MessageEndpoint> getEndpointClass() {
-		return SplitterEndpoint.class;
-	}
-
-	@Override
-	protected Class<?> getMethodInvokingAdapterClass() {
-		return MethodInvokingSplitter.class;
+	protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(SplitterEndpoint.class);
+		if (element.hasAttribute("ref")) {
+			String adapterBeanName = this.parseAdapter(element, parserContext, MethodInvokingSplitter.class);
+			builder.addConstructorArgReference(adapterBeanName);
+		}
+		return builder;
 	}
 
 }

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

@@ -16,7 +16,10 @@
 
 package org.springframework.integration.config;
 
-import org.springframework.integration.endpoint.MessageEndpoint;
+import org.w3c.dom.Element;
+
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.transformer.MethodInvokingTransformer;
 import org.springframework.integration.transformer.TransformerEndpoint;
 
@@ -28,13 +31,10 @@ import org.springframework.integration.transformer.TransformerEndpoint;
 public class TransformerParser extends AbstractEndpointParser {
 
 	@Override
-	protected Class<? extends MessageEndpoint> getEndpointClass() {
-		return TransformerEndpoint.class;
-	}
-
-	@Override
-	protected Class<?> getMethodInvokingAdapterClass() {
-		return MethodInvokingTransformer.class;
+	protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(TransformerEndpoint.class);
+		builder.addConstructorArgReference(this.parseAdapter(element, parserContext, MethodInvokingTransformer.class));
+		return builder;
 	}
 
 }

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

@@ -19,15 +19,21 @@ package org.springframework.integration.config.annotation;
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Method;
 
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.integration.annotation.Poller;
 import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.ChannelRegistry;
+import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PollableChannel;
-import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.channel.SubscribableChannel;
 import org.springframework.integration.endpoint.AbstractMessageConsumingEndpoint;
 import org.springframework.integration.endpoint.AbstractMessageHandlingEndpoint;
+import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.integration.endpoint.PollingConsumerEndpoint;
+import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.scheduling.IntervalTrigger;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
@@ -58,55 +64,73 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 	}
 
 	public Object postProcess(Object bean, String beanName, Method method, T annotation) {
-		Object adapter = this.createMethodInvokingAdapter(bean, method, annotation);
-		if (adapter != null && this.shouldCreateEndpoint(annotation)) {
-			AbstractEndpoint endpoint = this.createEndpoint(bean, adapter);
-			if (endpoint != null) {
-				Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class);
-				this.configureEndpoint(endpoint, annotation, pollerAnnotation);
-				endpoint.afterPropertiesSet();
-				return endpoint;
-			}
+		MessageConsumer consumer = this.createConsumer(bean, method, annotation);
+		if (consumer instanceof ChannelRegistryAware) {
+			((ChannelRegistryAware) consumer).setChannelRegistry(this.getChannelRegistry());
 		}
-		return adapter;
+		Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class);
+		MessageEndpoint endpoint = this.createEndpoint(consumer, annotation, pollerAnnotation);
+		if (endpoint != null) {
+			if (endpoint instanceof InitializingBean) {
+				try {
+					((InitializingBean) endpoint).afterPropertiesSet();
+				}
+				catch (Exception e) {
+					throw new IllegalStateException(
+							"failed to initialize annotation-based consumer", e);
+				}
+			}
+			return endpoint;
+		}
+		return consumer;
 	}
 
 	protected boolean shouldCreateEndpoint(T annotation) {
 		return (StringUtils.hasText((String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE)));
 	}
 
-	protected void configureEndpoint(AbstractEndpoint endpoint, T annotation, Poller pollerAnnotation) {
+	private MessageEndpoint createEndpoint(MessageConsumer consumer, T annotation, Poller pollerAnnotation) {
+		MessageEndpoint endpoint = null;
 		String inputChannelName = (String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE);
 		if (StringUtils.hasText(inputChannelName)) {
 			MessageChannel inputChannel = this.messageBus.lookupChannel(inputChannelName);
 			Assert.notNull(inputChannel, "unable to resolve inputChannel '" + inputChannelName + "'");
-			if (endpoint instanceof AbstractMessageConsumingEndpoint) {
-				AbstractMessageConsumingEndpoint consumingEndpoint = (AbstractMessageConsumingEndpoint) endpoint;
-				if (pollerAnnotation != null) {
-					Assert.isInstanceOf(PollableChannel.class, inputChannel,
-							"The @Poller annotation should only be provided for a PollableChannel");							
-					IntervalTrigger trigger = new IntervalTrigger(
-							pollerAnnotation.interval(), pollerAnnotation.timeUnit());
-					trigger.setInitialDelay(pollerAnnotation.initialDelay(), pollerAnnotation.timeUnit());
-					trigger.setFixedRate(pollerAnnotation.fixedRate());
-					consumingEndpoint.setTrigger(trigger);
-					consumingEndpoint.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
+			if (consumer instanceof AbstractMessageConsumingEndpoint) {
+				if (inputChannel instanceof PollableChannel) {
+					PollingConsumerEndpoint pollingEndpoint = new PollingConsumerEndpoint(
+							consumer, (PollableChannel) inputChannel);
+					if (pollerAnnotation != null) {
+						IntervalTrigger trigger = new IntervalTrigger(
+								pollerAnnotation.interval(), pollerAnnotation.timeUnit());
+						trigger.setInitialDelay(pollerAnnotation.initialDelay(), pollerAnnotation.timeUnit());
+						trigger.setFixedRate(pollerAnnotation.fixedRate());
+						pollingEndpoint.setTrigger(trigger);
+						pollingEndpoint.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
+					}
+					endpoint = pollingEndpoint;
+				}
+				else if (inputChannel instanceof SubscribableChannel) {
+					Assert.isTrue(pollerAnnotation == null,
+							"The @Poller annotation should only be provided for a PollableChannel");
+					endpoint = new SubscribingConsumerEndpoint(consumer, (SubscribableChannel) inputChannel);
+				}
+				else {
+					throw new IllegalArgumentException("unsupported channel type: ["
+							+ inputChannel.getClass() + "]");
 				}
-				consumingEndpoint.setInputChannel(inputChannel);
 			}
-			if (endpoint instanceof AbstractMessageHandlingEndpoint) {
+			if (consumer instanceof AbstractMessageHandlingEndpoint) {
 				String outputChannelName = (String) AnnotationUtils.getValue(annotation, OUTPUT_CHANNEL_ATTRIBUTE);
 				if (StringUtils.hasText(outputChannelName)) {
 					MessageChannel outputChannel = this.messageBus.lookupChannel(outputChannelName);
 					Assert.notNull(outputChannel, "unable to resolve outputChannel '" + outputChannelName + "'");
-					((AbstractMessageHandlingEndpoint) endpoint).setOutputChannel(outputChannel);
+					((AbstractMessageHandlingEndpoint) consumer).setOutputChannel(outputChannel);
 				}
 			}
 		}
+		return endpoint;
 	}
 
-	protected abstract Object createMethodInvokingAdapter(Object bean, Method method, T annotation);
-
-	protected abstract AbstractEndpoint createEndpoint(Object originalBean, Object adapter);
+	protected abstract MessageConsumer createConsumer(Object bean, Method method, T annotation);
 
 }

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

@@ -25,10 +25,9 @@ import org.springframework.integration.aggregator.CompletionStrategyAdapter;
 import org.springframework.integration.aggregator.MethodInvokingAggregator;
 import org.springframework.integration.annotation.Aggregator;
 import org.springframework.integration.annotation.CompletionStrategy;
-import org.springframework.integration.annotation.Poller;
 import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.util.Assert;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
@@ -46,44 +45,31 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
 
 
 	@Override
-	protected Object createMethodInvokingAdapter(Object bean, Method method, Aggregator annotation) {
-		return new MethodInvokingAggregator(bean, method);
-	}
-
-	@Override
-	protected AbstractEndpoint createEndpoint(Object originalBean, Object adapter) {
-		if (adapter instanceof org.springframework.integration.aggregator.Aggregator) {
-			AggregatorEndpoint endpoint = new AggregatorEndpoint((org.springframework.integration.aggregator.Aggregator) adapter);
-			this.configureCompletionStrategy(originalBean, endpoint);
-			return endpoint;
-		}
-		return null;
-	}
-
-	@Override
-	protected void configureEndpoint(AbstractEndpoint endpoint, Aggregator annotation, Poller pollerAnnotation) {
-		super.configureEndpoint(endpoint, annotation, pollerAnnotation);
-		AggregatorEndpoint aggregatorEndpoint = (AggregatorEndpoint) endpoint;
+	protected MessageConsumer createConsumer(Object bean, Method method, Aggregator annotation) {
+		MethodInvokingAggregator adapter = new MethodInvokingAggregator(bean, method);
+		AggregatorEndpoint aggregator = new AggregatorEndpoint(adapter);
+		this.configureCompletionStrategy(bean, aggregator);
 		String discardChannelName = annotation.discardChannel();
 		if (StringUtils.hasText(discardChannelName)) {
 			MessageChannel discardChannel = this.getChannelRegistry().lookupChannel(discardChannelName);
 			Assert.notNull(discardChannel, "unable to resolve discardChannel '" + discardChannelName + "'");
-			aggregatorEndpoint.setDiscardChannel(discardChannel);
+			aggregator.setDiscardChannel(discardChannel);
 		}
-		aggregatorEndpoint.setSendTimeout(annotation.sendTimeout());
-		aggregatorEndpoint.setSendPartialResultOnTimeout(annotation.sendPartialResultsOnTimeout());
-		aggregatorEndpoint.setReaperInterval(annotation.reaperInterval());
-		aggregatorEndpoint.setTimeout(annotation.timeout());
-		aggregatorEndpoint.setTrackedCorrelationIdCapacity(annotation.trackedCorrelationIdCapacity());
-		aggregatorEndpoint.afterPropertiesSet();
+		aggregator.setSendTimeout(annotation.sendTimeout());
+		aggregator.setSendPartialResultOnTimeout(annotation.sendPartialResultsOnTimeout());
+		aggregator.setReaperInterval(annotation.reaperInterval());
+		aggregator.setTimeout(annotation.timeout());
+		aggregator.setTrackedCorrelationIdCapacity(annotation.trackedCorrelationIdCapacity());
+		aggregator.afterPropertiesSet();
+		return aggregator;
 	}
 
-	private void configureCompletionStrategy(final Object object, final AggregatorEndpoint handler) {
-		ReflectionUtils.doWithMethods(object.getClass(), new ReflectionUtils.MethodCallback() {
+	private void configureCompletionStrategy(final Object bean, final AggregatorEndpoint aggregator) {
+		ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
 			public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
 				Annotation annotation = AnnotationUtils.getAnnotation(method, CompletionStrategy.class);
 				if (annotation != null) {
-					handler.setCompletionStrategy(new CompletionStrategyAdapter(object, method));
+					aggregator.setCompletionStrategy(new CompletionStrategyAdapter(bean, method));
 				}
 			}
 		});

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

@@ -26,9 +26,11 @@ 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.channel.SubscribableChannel;
 import org.springframework.integration.endpoint.MessageEndpoint;
-import org.springframework.integration.endpoint.OutboundChannelAdapter;
+import org.springframework.integration.endpoint.PollingConsumerEndpoint;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
 import org.springframework.integration.message.MethodInvokingConsumer;
 import org.springframework.integration.message.MethodInvokingSource;
 import org.springframework.integration.scheduling.IntervalTrigger;
@@ -102,16 +104,19 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
 		return adapter;
 	}
 
-	private OutboundChannelAdapter createOutboundChannelAdapter(MethodInvokingConsumer consumer, MessageChannel channel, Poller pollerAnnotation) {
-		OutboundChannelAdapter adapter = new OutboundChannelAdapter(consumer);
-		adapter.setInputChannel(channel);
+	private MessageEndpoint createOutboundChannelAdapter(MethodInvokingConsumer consumer, MessageChannel channel, Poller pollerAnnotation) {
 		if (channel instanceof PollableChannel) {
 			Trigger trigger = (pollerAnnotation != null)
 					? this.createTrigger(pollerAnnotation)
 					: new IntervalTrigger(0);
-			adapter.setTrigger(trigger);
+			PollingConsumerEndpoint endpoint = new PollingConsumerEndpoint(consumer, (PollableChannel) channel);
+			endpoint.setTrigger(trigger);
+			return endpoint;
 		}
-		return adapter;
+		if (channel instanceof SubscribableChannel) {
+			return new SubscribingConsumerEndpoint(consumer, (SubscribableChannel) channel);
+		}
+		return null;
 	}
 
 	private Trigger createTrigger(Poller pollerAnnotation) {

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

@@ -18,11 +18,10 @@ package org.springframework.integration.config.annotation;
 
 import java.lang.reflect.Method;
 
-import org.springframework.integration.annotation.Poller;
 import org.springframework.integration.annotation.Router;
 import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.router.MethodInvokingChannelResolver;
 import org.springframework.integration.router.RouterEndpoint;
 import org.springframework.util.Assert;
@@ -41,27 +40,16 @@ public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostP
 
 
 	@Override
-	protected Object createMethodInvokingAdapter(Object bean, Method method, Router annotation) {
-		return new MethodInvokingChannelResolver(bean, method);
-	}
-
-	@Override
-	protected AbstractEndpoint createEndpoint(Object originalBean, Object adapter) {
-		if (adapter instanceof MethodInvokingChannelResolver) {
-			return new RouterEndpoint((MethodInvokingChannelResolver) adapter);
-		}
-		return null;
-	}
-
-	@Override
-	protected void configureEndpoint(AbstractEndpoint endpoint, Router annotation, Poller pollerAnnotation) {
-		super.configureEndpoint(endpoint, annotation, pollerAnnotation);
+	protected MessageConsumer createConsumer(Object bean, Method method, Router annotation) {
+		MethodInvokingChannelResolver resolver = new MethodInvokingChannelResolver(bean, method);
+		RouterEndpoint router = new RouterEndpoint(resolver);
 		String defaultOutputChannelName = annotation.defaultOutputChannel();
 		if (StringUtils.hasText(defaultOutputChannelName)) {
 			MessageChannel defaultOutputChannel = this.getChannelRegistry().lookupChannel(defaultOutputChannelName);
 			Assert.notNull(defaultOutputChannel, "unable to resolve defaultOutputChannel '" + defaultOutputChannelName + "'");
-			((RouterEndpoint) endpoint).setDefaultOutputChannel(defaultOutputChannel);
+			router.setDefaultOutputChannel(defaultOutputChannel);
 		}
+		return router;
 	}
 
 }

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

@@ -20,10 +20,9 @@ import java.lang.reflect.Method;
 
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.bus.MessageBus;
-import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.message.MessageMappingMethodInvoker;
-import org.springframework.integration.util.MethodInvoker;
 
 /**
  * Post-processor for Methods annotated with {@link ServiceActivator @ServiceActivator}.
@@ -38,16 +37,9 @@ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnot
 
 
 	@Override
-	protected Object createMethodInvokingAdapter(Object bean, Method method, ServiceActivator annotation) {
-		return new MessageMappingMethodInvoker(bean, method);
-	}
-
-	@Override
-	protected AbstractEndpoint createEndpoint(Object originalBean, Object adapter) {
-		if (adapter instanceof MethodInvoker) {
-			return new ServiceActivatorEndpoint((MethodInvoker) adapter);
-		}
-		return null;
+	protected MessageConsumer createConsumer(Object bean, Method method, ServiceActivator annotation) {
+		MessageMappingMethodInvoker invoker = new MessageMappingMethodInvoker(bean, method);
+		return new ServiceActivatorEndpoint(invoker);
 	}
 
 }

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

@@ -20,7 +20,7 @@ import java.lang.reflect.Method;
 
 import org.springframework.integration.annotation.Splitter;
 import org.springframework.integration.bus.MessageBus;
-import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.splitter.MethodInvokingSplitter;
 import org.springframework.integration.splitter.SplitterEndpoint;
 
@@ -37,16 +37,9 @@ public class SplitterAnnotationPostProcessor extends AbstractMethodAnnotationPos
 
 
 	@Override
-	protected Object createMethodInvokingAdapter(Object bean, Method method, Splitter annotation) {
-		return new MethodInvokingSplitter(bean, method);
-	}
-
-	@Override
-	protected AbstractEndpoint createEndpoint(Object originalBean, Object adapter) {
-		if (adapter instanceof MethodInvokingSplitter) {
-			return new SplitterEndpoint((MethodInvokingSplitter) adapter);
-		}
-		return null;
+	protected MessageConsumer createConsumer(Object bean, Method method, Splitter annotation) {
+		MethodInvokingSplitter splitter = new MethodInvokingSplitter(bean, method);
+		return new SplitterEndpoint(splitter);
 	}
 
 }

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

@@ -20,7 +20,7 @@ import java.lang.reflect.Method;
 
 import org.springframework.integration.annotation.Transformer;
 import org.springframework.integration.bus.MessageBus;
-import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.transformer.MethodInvokingTransformer;
 import org.springframework.integration.transformer.TransformerEndpoint;
 
@@ -37,16 +37,9 @@ public class TransformerAnnotationPostProcessor extends AbstractMethodAnnotation
 
 
 	@Override
-	protected Object createMethodInvokingAdapter(Object bean, Method method, Transformer annotation) {
-		return new MethodInvokingTransformer(bean, method);
-	}
-
-	@Override
-	protected AbstractEndpoint createEndpoint(Object originalBean, Object adapter) {
-		if (adapter instanceof MethodInvokingTransformer) {
-			return new TransformerEndpoint((MethodInvokingTransformer) adapter);
-		}
-		return null;
+	protected MessageConsumer createConsumer(Object bean, Method method, Transformer annotation) {
+		MethodInvokingTransformer transformer = new MethodInvokingTransformer(bean, method);
+		return new TransformerEndpoint(transformer);
 	}
 
 }

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

@@ -16,65 +16,31 @@
 
 package org.springframework.integration.endpoint;
 
-import java.util.concurrent.ScheduledFuture;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
-import org.springframework.context.Lifecycle;
-import org.springframework.core.task.TaskExecutor;
-import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.channel.PollableChannel;
-import org.springframework.integration.channel.SubscribableChannel;
 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.scheduling.IntervalTrigger;
-import org.springframework.integration.scheduling.Trigger;
 import org.springframework.integration.util.ErrorHandler;
 import org.springframework.util.Assert;
 
 /**
- * The base class for Message Endpoint implementations that consume Messages.
+ * Base class for MessageConsumer implementations.
  * 
  * @author Mark Fisher
  */
-public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint implements MessageConsumer, Lifecycle {
+public abstract class AbstractMessageConsumingEndpoint implements MessageConsumer {
 
-	private volatile MessageChannel inputChannel;
-
-	private volatile Trigger trigger = new IntervalTrigger(0);
-
-	private volatile ChannelPoller poller;
-
-	private volatile ScheduledFuture<?> pollerFuture;
-
-	private volatile TaskExecutor taskExecutor;
-
-	private volatile int maxMessagesPerPoll = -1;
+	protected final Log logger = LogFactory.getLog(this.getClass());
 
 	private volatile ErrorHandler errorHandler;
 
-	private volatile boolean initialized;
-
-	private volatile boolean running;
-
-	private final Object lifecycleMonitor = new Object();
-
-
-	public void setInputChannel(MessageChannel inputChannel) {
-		this.inputChannel = inputChannel;
-	}
-
-	public void setTrigger(Trigger trigger) {
-		this.trigger = trigger;
-	}
-
-	public void setTaskExecutor(TaskExecutor taskExecutor) {
-		this.taskExecutor = taskExecutor;
-	}
 
 	/**
 	 * Provide an error handler for any Exceptions that occur
-	 * upon invocation of this endpoint. If none is provided,
+	 * upon invocation of this consumer. If none is provided,
 	 * the Exception messages will be logged (at warn level),
 	 * and the Exception rethrown.
 	 */
@@ -82,94 +48,11 @@ public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint
 		this.errorHandler = errorHandler;
 	}
 
-	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.trigger);
-				this.poller.setMaxMessagesPerPoll(this.maxMessagesPerPoll);
-				this.configureTransactionSettingsForPoller(this.poller);
-				if (this.taskExecutor != null) {
-					this.poller.setTaskExecutor(this.taskExecutor);
-				}
-				this.poller.setConsumer(this);
-			}
-			this.initialized = true;
-		}
-	}
-
-	public final void start() {
-		synchronized (this.lifecycleMonitor) {
-			if (this.running) {
-				return;
-			}
-			if (!this.initialized) {
-				this.afterPropertiesSet();
-			}
-			Assert.notNull(this.inputChannel, "failed to start endpoint, inputChannel is required");
-			if (this.inputChannel instanceof SubscribableChannel) {
-				((SubscribableChannel) inputChannel).subscribe(this);
-			}
-			else if (this.inputChannel instanceof PollableChannel) {
-				Assert.notNull(this.getTaskScheduler(),
-						"failed to start endpoint, no taskScheduler available");
-				this.pollerFuture = this.getTaskScheduler().schedule(this.poller, this.poller.getTrigger());
-			}
-			onStart();
-			this.running = true;
-		}
-	}
-
-	public final void stop() {
-		synchronized (this.lifecycleMonitor) {
-			if (!this.running) {
-				return;
-			}
-			if (this.inputChannel instanceof SubscribableChannel) {
-				((SubscribableChannel) inputChannel).unsubscribe(this);
-			}
-			else if (this.pollerFuture != null) {
-				this.pollerFuture.cancel(true);
-			}
-			onStop();
-			this.running = false;
-		}
-	}
-	
-	/**
-	 * Subclasses might override this to supply their own start code (e.g. if they start threads
-	 * on their own). This method will be called within the lifecycleMonitor.
-	 */
-	protected void onStart() {
-		
-	}
-	
-	/**
-	 * Subclasses might override this to supply their own stop code (e.g. if they stop threads
-	 * on their own).This method will be called within the lifecycleMonitor.
-	 * 
-	 */
-	protected void onStop() {
-		
-	}
-
 	public final void onMessage(Message<?> message) {
-		if (message == null || message.getPayload() == null) {
-			throw new IllegalArgumentException("Message and its payload must not be null");
-		}
+		Assert.notNull(message == null, "Message must not be null");
+		Assert.notNull(message.getPayload(), "Message payload must not be null");
 		if (this.logger.isDebugEnabled()) {
-			this.logger.debug("endpoint '" + this + "' processing message: " + message);
+			this.logger.debug("consumer '" + this + "' processing message: " + message);
 		}
 		try {
 			this.onMessageInternal(message);
@@ -180,7 +63,7 @@ public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint
 			}
 			else {
 				this.handleException(new MessageHandlingException(message,
-						"failure occurred in endpoint '" + this.toString() + "'", e));
+						"failure occurred in consumer '" + this.toString() + "'", e));
 			}
 		}
 	}
@@ -188,7 +71,7 @@ public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint
 	protected void handleException(MessagingException exception) {
 		if (this.errorHandler == null) {
 			if (this.logger.isWarnEnabled()) {
-				this.logger.warn("exception occurred in endpoint '" + this + "'", exception);
+				this.logger.warn("exception occurred in consumer '" + this + "'", exception);
 			}
 			throw exception;
 		}

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

@@ -22,6 +22,7 @@ import java.util.List;
 import org.springframework.integration.channel.ChannelRegistry;
 import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.MessageChannelTemplate;
 import org.springframework.integration.message.CompositeMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageBuilder;
@@ -30,6 +31,7 @@ import org.springframework.integration.message.MessageHeaders;
 import org.springframework.integration.message.MessageRejectedException;
 import org.springframework.integration.message.MessagingException;
 import org.springframework.integration.message.selector.MessageSelector;
+import org.springframework.util.Assert;
 
 /**
  * @author Mark Fisher
@@ -44,6 +46,8 @@ public abstract class AbstractMessageHandlingEndpoint extends AbstractMessageCon
 
 	private volatile boolean requiresReply = false;
 
+	private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
+
 
 	public void setOutputChannel(MessageChannel outputChannel) {
 		this.outputChannel = outputChannel;
@@ -74,7 +78,7 @@ public abstract class AbstractMessageHandlingEndpoint extends AbstractMessageCon
 		Object result = this.handle(message);
 		if (result == null) {
 			if (this.requiresReply) {
-				throw new MessageHandlingException(message, "endpoint '" + this
+				throw new MessageHandlingException(message, "consumer '" + this
 						+ "' requires a reply, but no reply was received");
 			}
 			return;
@@ -105,7 +109,7 @@ public abstract class AbstractMessageHandlingEndpoint extends AbstractMessageCon
 	protected boolean supports(Message<?> message) {
 		if (this.selector != null && !this.selector.accept(message)) {
 			if (logger.isDebugEnabled()) {
-				logger.debug("selector for endpoint '" + this + "' rejected message: " + message);
+				logger.debug("selector for consumer '" + this + "' rejected message: " + message);
 			}
 			return false;
 		}
@@ -117,7 +121,7 @@ public abstract class AbstractMessageHandlingEndpoint extends AbstractMessageCon
 	}
 
 	private boolean sendReplyMessage(Message<?> replyMessage, MessageChannel replyChannel) {
-		return this.getChannelTemplate().send(replyMessage, replyChannel);
+		return this.channelTemplate.send(replyMessage, replyChannel);
 	}
 
 	private Message<?> buildReplyMessage(Object result, MessageHeaders requestHeaders) {
@@ -153,9 +157,9 @@ public abstract class AbstractMessageHandlingEndpoint extends AbstractMessageCon
 					replyChannel = (MessageChannel) returnAddress;
 				}
 				else if (returnAddress instanceof String) {
-					if (this.channelRegistry != null) {
-						replyChannel = this.channelRegistry.lookupChannel((String) returnAddress);
-					}
+					Assert.state(this.channelRegistry != null,
+							"ChannelRegistry is required for resolving a reply channel by name");
+					replyChannel = this.channelRegistry.lookupChannel((String) returnAddress);
 				}
 				else {
 					throw new MessagingException("expected a MessageChannel or String for 'returnAddress', but type is ["

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

@@ -140,6 +140,9 @@ public abstract class AbstractPollingEndpoint implements MessageEndpoint, TaskSc
 			if (!this.initialized) {
 				this.afterPropertiesSet();
 			}
+			if (this.isRunning()) {
+				return;
+			}
 			Assert.state(this.taskScheduler != null,
 					"unable to start polling, no taskScheduler available");
 			this.runningTask = this.taskScheduler.schedule(new Poller(), this.trigger);
@@ -176,7 +179,7 @@ public abstract class AbstractPollingEndpoint implements MessageEndpoint, TaskSc
 
 		private void poll() {
 			int count = 0;
-			while (maxMessagesPerPoll < 0 || count < maxMessagesPerPoll) {
+			while (maxMessagesPerPoll <= 0 || count < maxMessagesPerPoll) {
 				if (!innerPoll()) {
 					break;
 				}

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

@@ -31,7 +31,7 @@ import org.springframework.util.Assert;
 /**
  * @author Mark Fisher
  */
-public class ServiceActivatorEndpoint extends AbstractMessageHandlingEndpoint {
+public class ServiceActivatorEndpoint extends AbstractMessageHandlingEndpoint implements InitializingBean {
 
 	private final MethodResolver methodResolver = new DefaultMethodResolver(ServiceActivator.class);
 
@@ -56,9 +56,7 @@ public class ServiceActivatorEndpoint extends AbstractMessageHandlingEndpoint {
 	}
 
 
-	@Override
-	protected void initialize() throws Exception {
-		super.initialize();
+	public void afterPropertiesSet() throws Exception {
 		if (this.invoker instanceof InitializingBean) {
 			((InitializingBean) this.invoker).afterPropertiesSet();
 		}

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/config/ConsumerEndpointFactoryBean.java

@@ -22,6 +22,8 @@ import org.springframework.beans.factory.FactoryBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.core.task.TaskExecutor;
+import org.springframework.integration.channel.ChannelRegistry;
+import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PollableChannel;
@@ -39,7 +41,7 @@ import org.springframework.util.Assert;
 /**
  * @author Mark Fisher
  */
-public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAware, InitializingBean {
+public class ConsumerEndpointFactoryBean implements FactoryBean, ChannelRegistryAware, BeanFactoryAware, InitializingBean {
 
 	private final MessageConsumer consumer;
 
@@ -76,6 +78,16 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAwar
 		this.inputChannelName = inputChannelName;
 	}
 
+	public void setChannelRegistry(ChannelRegistry channelRegistry) {
+		if (this.consumer instanceof ChannelRegistryAware) {
+			((ChannelRegistryAware) this.consumer).setChannelRegistry(channelRegistry);
+		}
+	}
+
+	public void setTrigger(Trigger trigger) {
+		this.trigger = trigger;
+	}
+
 	public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
 		this.maxMessagesPerPoll = maxMessagesPerPoll;
 	}

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

@@ -21,8 +21,14 @@ import org.springframework.integration.bus.MessageBusAware;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.MessageChannelTemplate;
 import org.springframework.integration.channel.PollableChannel;
+import org.springframework.integration.channel.SubscribableChannel;
+import org.springframework.integration.endpoint.AbstractMessageHandlingEndpoint;
+import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.endpoint.MessagingGateway;
+import org.springframework.integration.endpoint.PollingConsumerEndpoint;
+import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
 import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.message.MessageDeliveryException;
 import org.springframework.util.Assert;
 
@@ -42,7 +48,7 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
 
 	private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
 
-	private volatile ReplyMessageCorrelator replyMessageCorrelator;
+	private volatile MessageEndpoint replyMessageCorrelator;
 
 	private volatile MessageBus messageBus;
 
@@ -145,10 +151,23 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
 				return;
 			}
 			Assert.state(this.messageBus != null, "No MessageBus available. Cannot register ReplyMessageCorrelator.");
-			ReplyMessageCorrelator correlator = new ReplyMessageCorrelator();
-			correlator.setBeanName("internal.correlator." + this);
-			correlator.setInputChannel(this.replyChannel);
-			correlator.afterPropertiesSet();
+			MessageEndpoint correlator = null;
+			MessageConsumer consumer = new AbstractMessageHandlingEndpoint() {
+				@Override
+				protected Object handle(Message<?> message) {
+					return message;
+				}
+			};
+			if (this.replyChannel instanceof SubscribableChannel) {
+				correlator = new SubscribingConsumerEndpoint(
+						consumer, (SubscribableChannel) this.replyChannel);
+			}
+			else if (this.replyChannel instanceof PollableChannel) {
+				PollingConsumerEndpoint endpoint = new PollingConsumerEndpoint(
+						consumer, (PollableChannel) this.replyChannel);
+				endpoint.afterPropertiesSet();
+				correlator = endpoint;
+			}
 			this.messageBus.registerEndpoint(correlator);
 			this.replyMessageCorrelator = correlator;
 		}

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

@@ -21,6 +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.channel.MessageChannelTemplate;
 import org.springframework.integration.endpoint.AbstractMessageConsumingEndpoint;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageDeliveryException;
@@ -37,6 +38,8 @@ public class RouterEndpoint extends AbstractMessageConsumingEndpoint implements
 
 	private volatile boolean resolutionRequired;
 
+	private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
+
 
 	public RouterEndpoint(ChannelResolver channelResolver) {
 		Assert.notNull(channelResolver, "ChannelResolver must not be null");
@@ -59,7 +62,7 @@ public class RouterEndpoint extends AbstractMessageConsumingEndpoint implements
 	 * default, there is no timeout, meaning the send will block indefinitely.
 	 */
 	public void setTimeout(long timeout) {
-		this.getChannelTemplate().setSendTimeout(timeout);
+		this.channelTemplate.setSendTimeout(timeout);
 	}
 
 	/**
@@ -79,7 +82,7 @@ public class RouterEndpoint extends AbstractMessageConsumingEndpoint implements
 		if (results != null) {
 			for (MessageChannel channel : results) {
 				if (channel != null) {
-					if (this.getChannelTemplate().send(message, channel)) {
+					if (this.channelTemplate.send(message, channel)) {
 						sent = true;
 					}
 				}
@@ -87,7 +90,7 @@ public class RouterEndpoint extends AbstractMessageConsumingEndpoint implements
 		}
 		if (!sent) {
 			if (this.defaultOutputChannel != null) {
-				sent = this.getChannelTemplate().send(message, this.defaultOutputChannel);
+				sent = this.channelTemplate.send(message, this.defaultOutputChannel);
 			}
 			else if (this.resolutionRequired) {
 				throw new MessageDeliveryException(message,

org.springframework.integration/src/main/java/org/springframework/integration/transformer/config/AbstractTransformerParser.java

@@ -22,7 +22,6 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
 import org.springframework.beans.factory.xml.ParserContext;
 import org.springframework.integration.config.AbstractEndpointParser;
-import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.transformer.Transformer;
 import org.springframework.integration.transformer.TransformerEndpoint;
 
@@ -32,23 +31,15 @@ import org.springframework.integration.transformer.TransformerEndpoint;
 public abstract class AbstractTransformerParser extends AbstractEndpointParser {
 
 	@Override
-	protected boolean shouldCreateAdapter(Element element) {
-		return false;
-	}
-
-	@Override
-	protected Class<? extends MessageEndpoint> getEndpointClass() {
-		return TransformerEndpoint.class;
-	}
-
-	@Override
-	protected void postProcess(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+	protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(TransformerEndpoint.class);
 		BeanDefinitionBuilder transformerBuilder =
 				BeanDefinitionBuilder.genericBeanDefinition(this.getTransformerClass());
 		this.parseTransformer(element, parserContext, transformerBuilder);
 		String transformerBeanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
 				transformerBuilder.getBeanDefinition(), parserContext.getRegistry());
 		builder.addConstructorArgReference(transformerBeanName);
+		return builder;
 	}
 
 	protected abstract Class<? extends Transformer> getTransformerClass();