diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java
index 88d2f7340f..b8221ea1cc 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java
@@ -30,11 +30,12 @@ import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
-import org.springframework.integration.endpoint.ReplyMessageHolder;
+import org.springframework.integration.channel.MessageChannelTemplate;
+import org.springframework.integration.endpoint.AbstractMessageConsumer;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageConsumer;
 import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.integration.message.MessageProducer;
 import org.springframework.integration.scheduling.IntervalTrigger;
 import org.springframework.integration.scheduling.TaskScheduler;
 import org.springframework.integration.scheduling.TaskSchedulerAware;
@@ -64,8 +65,10 @@ import org.springframework.util.ObjectUtils;
  * @author Mark Fisher
  * @author Marius Bogoevici
  */
-public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProducingMessageConsumer
-		implements TaskSchedulerAware, InitializingBean {
+public abstract class AbstractMessageBarrierConsumer extends AbstractMessageConsumer
+		implements MessageProducer, TaskSchedulerAware, InitializingBean {
+
+	public final static long DEFAULT_SEND_TIMEOUT = 1000;
 
 	public final static long DEFAULT_TIMEOUT = 60000;
 
@@ -73,8 +76,13 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
 
 	public final static int DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY = 1000;
 
+
 	protected final Log logger = LogFactory.getLog(this.getClass());
 
+	private MessageChannel outputChannel;
+
+	private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
+
 	private volatile MessageChannel discardChannel;
 
 	protected final ConcurrentMap<Object, MessageBarrier> barriers =
@@ -97,6 +105,15 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
 	private ScheduledFuture<?> reaperFutureTask;
 
 
+	public AbstractMessageBarrierConsumer() {
+		this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
+	}
+
+
+	public void setOutputChannel(MessageChannel outputChannel) {
+		this.outputChannel = outputChannel;
+	}
+
 	/**
 	 * Specify a channel for sending Messages that arrive after their aggregation
 	 * group has either completed or timed-out.
@@ -138,6 +155,10 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
 		this.timeout = timeout;
 	}
 
+	public void setSendTimeout(long sendTimeout) {
+		this.channelTemplate.setSendTimeout(sendTimeout);
+	}
+
 	public void setTaskScheduler(TaskScheduler taskScheduler) {
 		this.taskScheduler = taskScheduler;
 	}
@@ -167,7 +188,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
 	}
 
 	@Override
-	protected final void onMessage(Message<?> message, ReplyMessageHolder replyHolder) {
+	protected final void onMessageInternal(Message<?> message) {
 		if (!this.initialized) {
 			this.afterPropertiesSet();
 		}
@@ -177,36 +198,47 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
 					this.getClass().getSimpleName() + " requires the 'correlationId' property");
 		}
 		if (this.trackedCorrelationIds.contains(correlationId)) {
-			if (logger.isDebugEnabled()) {
-				logger.debug("Handling of Message group with correlationId '"
-						+ correlationId + "' has already completed or timed out.");
-			}
-			this.sendToDiscardChannelIfAvailable(message);
-			return;
+			this.discardMessage(message, correlationId);
 		}
+		else {
+			this.processMessage(message, correlationId);
+		}
+	}
+
+	private void discardMessage(Message<?> message, Object correlationId) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Handling of Message group with correlationId '"
+					+ correlationId + "' has already completed or timed out.");
+		}
+		if (this.discardChannel != null) {
+			boolean sent = this.channelTemplate.send(message, this.discardChannel);
+			if (!sent && logger.isWarnEnabled()) {
+				logger.warn("unable to send to 'discardChannel', message: " + message);
+			}
+		}
+	}
+
+	private void processMessage(Message<?> message, Object correlationId) {
 		MessageBarrier barrier = barriers.putIfAbsent(correlationId, createMessageBarrier());
 		if (barrier == null) {
 			barrier = barriers.get(correlationId);
 		}
 		List<Message<?>> releasedMessages = barrier.addAndRelease(message);
-		if (CollectionUtils.isEmpty(releasedMessages)) {
-			return;
+		if (!CollectionUtils.isEmpty(releasedMessages)) {
+			if (isBarrierRemovable(correlationId, releasedMessages)) {
+				this.removeBarrier(correlationId);
+			}
+			Message<?>[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages);
+			if (!ObjectUtils.isEmpty(processedMessages)) {
+				this.afterRelease(correlationId, releasedMessages);
+			}
 		}
-		if (isBarrierRemovable(correlationId, releasedMessages)) {
-			this.removeBarrier(correlationId);
-		}
-		Message<?>[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages);
-		if (ObjectUtils.isEmpty(processedMessages)) {
-			return;
-		}
-		this.afterRelease(correlationId, releasedMessages);
-		return;
 	}
 
 	private void afterRelease(Object correlationId, List<Message<?>> releasedMessages) {
 		Message<?>[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages);
 		for (Message<?> result : processedMessages) {
-			MessageChannel replyChannel = this.getOutputChannel();
+			MessageChannel replyChannel = this.outputChannel;
 			if (replyChannel == null) {
 				replyChannel = this.resolveReplyChannelFromMessage(result);
 				if (replyChannel == null) {
@@ -214,7 +246,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
 				}
 			}
 			if (replyChannel != null) {
-				this.sendReplyMessage(result, replyChannel);
+				this.channelTemplate.send(result, replyChannel);
 			}
 			else if (logger.isWarnEnabled()) {
 				logger.warn("unable to determine reply target for aggregation result: " + result);
@@ -222,17 +254,6 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
 		}
 	}
 
-	private void sendToDiscardChannelIfAvailable(Message<?> message) {
-		if (this.discardChannel != null) {
-			boolean sent = this.sendReplyMessage(message, this.discardChannel);
-			if (!sent) {
-				if (logger.isWarnEnabled()) {
-					logger.warn("unable to send to 'discardChannel', message: " + message);
-				}
-			}
-		}
-	}
-
 	protected MessageChannel resolveReplyChannelFromMessage(Message<?> message) {
 		Object returnAddress = message.getHeaders().getReturnAddress();
 		if (returnAddress != null) {
@@ -273,7 +294,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc
 					}
 					else {
 						for (Message<?> message : messages) {
-							sendToDiscardChannelIfAvailable(message);
+							discardMessage(message, correlationId);
 						}
 					}
 				}
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
index 2341be9640..aacd2c0766 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
@@ -19,6 +19,7 @@ package org.springframework.integration.config.annotation;
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Method;
 
+import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.generic.GenericBeanFactoryAccessor;
@@ -29,12 +30,11 @@ import org.springframework.integration.channel.ChannelResolver;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.channel.SubscribableChannel;
-import org.springframework.integration.endpoint.AbstractMessageConsumer;
-import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer;
 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.message.MessageProducer;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -91,34 +91,33 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 		if (StringUtils.hasText(inputChannelName)) {
 			MessageChannel inputChannel = this.channelResolver.resolveChannelName(inputChannelName);
 			Assert.notNull(inputChannel, "failed to resolve inputChannel '" + inputChannelName + "'");
-			if (consumer instanceof AbstractMessageConsumer) {
-				if (inputChannel instanceof PollableChannel) {
-					PollingConsumerEndpoint pollingEndpoint = new PollingConsumerEndpoint(
-							consumer, (PollableChannel) inputChannel);
-					if (pollerAnnotation != null) {
-						AnnotationConfigUtils.configurePollingEndpointWithPollerAnnotation(
-								pollingEndpoint, pollerAnnotation, this.beanFactoryAccessor.getBeanFactory());
-					}
-					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() + "]");
+			if (inputChannel instanceof PollableChannel) {
+				PollingConsumerEndpoint pollingEndpoint = new PollingConsumerEndpoint(
+						consumer, (PollableChannel) inputChannel);
+				if (pollerAnnotation != null) {
+					AnnotationConfigUtils.configurePollingEndpointWithPollerAnnotation(
+							pollingEndpoint, pollerAnnotation, this.beanFactoryAccessor.getBeanFactory());
 				}
+				endpoint = pollingEndpoint;
 			}
-			if (consumer instanceof AbstractReplyProducingMessageConsumer) {
+			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() + "]");
+			}
+			if (consumer instanceof MessageProducer) {
 				String outputChannelName = (String) AnnotationUtils.getValue(annotation, OUTPUT_CHANNEL_ATTRIBUTE);
 				if (StringUtils.hasText(outputChannelName)) {
 					MessageChannel outputChannel = this.channelResolver.resolveChannelName(outputChannelName);
 					Assert.notNull(outputChannel, "unable to resolve outputChannel '" + outputChannelName + "'");
-					((AbstractReplyProducingMessageConsumer) consumer).setOutputChannel(outputChannel);
+					((MessageProducer) consumer).setOutputChannel(outputChannel);
 				}
-				((AbstractReplyProducingMessageConsumer) consumer).setChannelResolver(this.channelResolver);
+			}
+			if (consumer instanceof BeanFactoryAware) {
+				((BeanFactoryAware) consumer).setBeanFactory(this.beanFactoryAccessor.getBeanFactory());
 			}
 		}
 		return endpoint;
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java
index ab5f7b92ab..2a4520feff 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java
@@ -26,6 +26,7 @@ import org.springframework.integration.message.Message;
 import org.springframework.integration.message.MessageBuilder;
 import org.springframework.integration.message.MessageHandlingException;
 import org.springframework.integration.message.MessageHeaders;
+import org.springframework.integration.message.MessageProducer;
 import org.springframework.integration.message.MessageRejectedException;
 import org.springframework.integration.message.MessagingException;
 import org.springframework.integration.message.selector.MessageSelector;
@@ -36,7 +37,8 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public abstract class AbstractReplyProducingMessageConsumer extends AbstractMessageConsumer implements BeanFactoryAware {
+public abstract class AbstractReplyProducingMessageConsumer extends AbstractMessageConsumer
+		implements MessageProducer, BeanFactoryAware {
 
 	public static final long DEFAULT_SEND_TIMEOUT = 1000;
 
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageProducer.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageProducer.java
new file mode 100644
index 0000000000..950619afb5
--- /dev/null
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageProducer.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.message;
+
+import org.springframework.integration.channel.MessageChannel;
+
+/**
+ * Base interface for any component that produces Messages.
+ * 
+ * @author Mark Fisher
+ */
+public interface MessageProducer {
+
+	void setOutputChannel(MessageChannel outputChannel);
+
+}