diff --git a/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractPollableConsumerTestBinder.java b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractPollableConsumerTestBinder.java
new file mode 100644
index 000000000..ea34dcb16
--- /dev/null
+++ b/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractPollableConsumerTestBinder.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2018 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.cloud.stream.binder;
+
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public abstract class AbstractPollableConsumerTestBinder<C extends AbstractBinder<MessageChannel, CP, PP>,
+	CP extends ConsumerProperties, PP extends ProducerProperties> extends AbstractTestBinder<C, CP, PP>
+	implements PollableConsumerBinder<MessageHandler, CP>{
+
+	private PollableConsumerBinder<MessageHandler, CP> binder;
+
+	@SuppressWarnings("unchecked")
+	public void setPollableConsumerBinder(PollableConsumerBinder<MessageHandler, CP> binder) {
+		super.setBinder((C) binder);
+		this.binder = binder;
+	}
+
+	@Override
+	public Binding<PollableSource<MessageHandler>> bindPollableConsumer(String name, String group,
+			PollableSource<MessageHandler> inboundBindTarget, CP consumerProperties) {
+		return this.binder.bindPollableConsumer(name, group, inboundBindTarget, consumerProperties);
+	}
+
+}
diff --git a/spring-cloud-stream/pom.xml b/spring-cloud-stream/pom.xml
index 3b5610886..d8298ad2c 100644
--- a/spring-cloud-stream/pom.xml
+++ b/spring-cloud-stream/pom.xml
@@ -33,6 +33,7 @@
 		<dependency>
 			<groupId>org.springframework.integration</groupId>
 			<artifactId>spring-integration-core</artifactId>
+			<version>5.0.1.BUILD-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.integration</groupId>
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
index 71d32021b..c0c839728 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -31,6 +31,7 @@ import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.handler.BridgeHandler;
 import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
@@ -41,6 +42,7 @@ import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.support.ChannelInterceptorAdapter;
+import org.springframework.retry.RecoveryCallback;
 import org.springframework.util.Assert;
 
 /**
@@ -60,11 +62,12 @@ import org.springframework.util.Assert;
  * @author Soby Chacko
  * @author Oleg Zhurakousky
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 1.1
  */
 public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties, P extends ProducerProperties, PP extends ProvisioningProvider<C, P>>
-		extends AbstractBinder<MessageChannel, C, P> {
+		extends AbstractBinder<MessageChannel, C, P> implements PollableConsumerBinder<MessageHandler, C> {
 
 	private final EmbeddedHeadersChannelInterceptor embeddedHeadersChannelInterceptor =
 			new EmbeddedHeadersChannelInterceptor(this.logger);
@@ -286,6 +289,65 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 	}
 
+
+	@Override
+	public Binding<PollableSource<MessageHandler>> bindPollableConsumer(String name, String group,
+			final PollableSource<MessageHandler> inboundBindTarget, C properties) {
+		Assert.isInstanceOf(DefaultPollableMessageSource.class, inboundBindTarget);
+		DefaultPollableMessageSource bindingTarget = (DefaultPollableMessageSource) inboundBindTarget;
+		ConsumerDestination destination = this.provisioningProvider.provisionConsumerDestination(name, group,
+				properties);
+		if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) {
+			bindingTarget.addInterceptor(0, this.embeddedHeadersChannelInterceptor);
+		}
+		final PolledConsumerResources resources = createPolledConsumerResources(name, group, destination, properties);
+		bindingTarget.setSource(resources.getSource());
+		if (resources.getErrorInfrastructure() != null) {
+			if (resources.getErrorInfrastructure().getErrorChannel() != null) {
+				bindingTarget.setErrorChannel(resources.getErrorInfrastructure().getErrorChannel());
+			}
+			ErrorMessageStrategy ems = getErrorMessageStrategy();
+			if (ems != null) {
+				bindingTarget.setErrorMessageStrategy(ems);
+			}
+		}
+		if (properties.getMaxAttempts() > 1) {
+			bindingTarget.setRetryTemplate(buildRetryTemplate(properties));
+			bindingTarget.setRecoveryCallback(
+					getPolledConsumerRecoveryCallback(resources.getErrorInfrastructure(), properties));
+		}
+		postProcessPollableSource(bindingTarget);
+		return new DefaultBinding<PollableSource<MessageHandler>>(name, group, inboundBindTarget,
+				resources.getSource() instanceof Lifecycle ? (Lifecycle) resources.getSource() : null) {
+
+			@Override
+			public void afterUnbind() {
+				afterUnbindConsumer(destination, this.group, properties);
+				destroyErrorInfrastructure(destination, group, properties);
+			}
+
+		};
+	}
+
+	protected void postProcessPollableSource(DefaultPollableMessageSource bindingTarget) {
+	}
+
+	/**
+	 * Implementations can override the default {@link ErrorMessageSendingRecoverer}.
+	 * @param errorInfrastructure the infrastructure.
+	 * @param properties the consumer properties.
+	 * @return the recoverer.
+	 */
+	protected RecoveryCallback<Object> getPolledConsumerRecoveryCallback(ErrorInfrastructure errorInfrastructure,
+			C properties) {
+		return errorInfrastructure.getRecoverer();
+	}
+
+	protected PolledConsumerResources createPolledConsumerResources(String name, String group,
+			ConsumerDestination destination, C consumerProperties) {
+		throw new UnsupportedOperationException("This binder does not support pollable consumers");
+	}
+
 	private void enhanceMessageChannel(MessageChannel inputChannel) {
 		((AbstractMessageChannel) inputChannel).addInterceptor(0, this.embeddedHeadersChannelInterceptor);
 	}
@@ -363,6 +425,23 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	 */
 	protected final ErrorInfrastructure registerErrorInfrastructure(ConsumerDestination destination, String group,
 			C consumerProperties) {
+
+		return registerErrorInfrastructure(destination, group, consumerProperties, false);
+	}
+
+	/**
+	 * Build an errorChannelRecoverer that writes to a pub/sub channel for the destination
+	 * when an exception is thrown to a consumer.
+	 * @param destination the destination.
+	 * @param group the group.
+	 * @param consumerProperties the properties.
+	 * @param true if this is for a polled consumer.
+	 * @return the ErrorInfrastructure which is a holder for the error channel, the recoverer and the
+	 * message handler that is subscribed to the channel.
+	 */
+	protected final ErrorInfrastructure registerErrorInfrastructure(ConsumerDestination destination, String group,
+			C consumerProperties, boolean polled) {
+
 		ErrorMessageStrategy errorMessageStrategy = getErrorMessageStrategy();
 		ConfigurableListableBeanFactory beanFactory = getApplicationContext().getBeanFactory();
 		String errorChannelName = errorsBaseName(destination, group, consumerProperties);
@@ -390,7 +469,13 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		String recovererBeanName = getErrorRecovererName(destination, group, consumerProperties);
 		beanFactory.registerSingleton(recovererBeanName, recoverer);
 		beanFactory.initializeBean(recoverer, recovererBeanName);
-		MessageHandler handler = getErrorMessageHandler(destination, group, consumerProperties);
+		MessageHandler handler;
+		if (polled) {
+			handler = getPolledConsumerErrorMessageHandler(destination, group, consumerProperties);
+		}
+		else {
+			handler = getErrorMessageHandler(destination, group, consumerProperties);
+		}
 		MessageChannel defaultErrorChannel = null;
 		if (getApplicationContext().containsBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)) {
 			defaultErrorChannel = getApplicationContext().getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME,
@@ -482,8 +567,22 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	 * @return the handler (may be null, which is the default, causing the exception to be
 	 * rethrown).
 	 */
-	protected MessageHandler getErrorMessageHandler(final ConsumerDestination destination, final String group,
-			final C consumerProperties) {
+	protected MessageHandler getErrorMessageHandler(ConsumerDestination destination, String group,
+			C consumerProperties) {
+		return null;
+	}
+
+	/**
+	 * Binders can return a message handler to be subscribed to the error channel.
+	 * Examples might be if the user wishes to (re)publish messages to a DLQ.
+	 * @param destination the destination.
+	 * @param group the group.
+	 * @param consumerProperties the properties.
+	 * @return the handler (may be null, which is the default, causing the exception to be
+	 * rethrown).
+	 */
+	protected MessageHandler getPolledConsumerErrorMessageHandler(ConsumerDestination destination, String group,
+			C consumerProperties) {
 		return null;
 	}
 
@@ -674,4 +773,25 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 
 	}
 
+	protected static class PolledConsumerResources {
+
+		private final MessageSource<?> source;
+
+		private final ErrorInfrastructure errorInfrastructure;
+
+		public PolledConsumerResources(MessageSource<?> source, ErrorInfrastructure errorInfrastructure) {
+			this.source = source;
+			this.errorInfrastructure = errorInfrastructure;
+		}
+
+		MessageSource<?> getSource() {
+			return this.source;
+		}
+
+		ErrorInfrastructure getErrorInfrastructure() {
+			return this.errorInfrastructure;
+		}
+
+	}
+
 }
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java
index 6461523ce..fe82232f1 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2016 the original author or authors.
+ * Copyright 2013-2018 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.
@@ -21,6 +21,10 @@ package org.springframework.cloud.stream.binder;
  * intended to identify a logical consumer or producer of messages. This may be a queue, a
  * channel adapter, another message channel, a Spring bean, etc.
  *
+ * @param <T> the primary binding type (e.g. MessageChannel).
+ * @param <C> the consumer properties type.
+ * @param <P> the producer properties type.
+ *
  * @author Mark Fisher
  * @author David Turanski
  * @author Gary Russell
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
index 07809f103..add17fe17 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -47,6 +47,7 @@ import org.springframework.util.StringUtils;
  *
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
+ * @author Gary Russell
  */
 public class DefaultBinderFactory implements BinderFactory, DisposableBean, ApplicationContextAware {
 
@@ -136,11 +137,27 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
 			configurationName = name;
 		}
 		Binder<T, ?, ?> binderInstance = getBinderInstance(configurationName);
-		Assert.state(GenericsUtils.getParameterType(binderInstance.getClass(), Binder.class, 0).isAssignableFrom(bindingTargetType),
+		Assert.state(verifyBinderTypeMatchesTarget(binderInstance, bindingTargetType),
 				"The binder '" + configurationName + "' cannot bind a " + bindingTargetType.getName());
 		return binderInstance;
 	}
 
+	/**
+	 * Return true if the binder is a {@link PollableConsumerBinder} and the target type
+	 * is a {@link PollableSource} and their generic types match (e.g. MessageHandler), OR
+	 * if it's a {@link Binder} and the target matches the binder's generic type.
+	 * @param binderInstance the binder.
+	 * @param bindingTargetType the binding target type.
+	 * @return true if the conditions match.
+	 */
+	private <T> boolean verifyBinderTypeMatchesTarget(Binder<T, ?, ?> binderInstance,
+			Class<? extends T> bindingTargetType) {
+		return (binderInstance instanceof PollableConsumerBinder
+				&& GenericsUtils.checkCompatiblePollableBinder(binderInstance, bindingTargetType))
+			|| GenericsUtils.getParameterType(binderInstance.getClass(), Binder.class, 0)
+					.isAssignableFrom(bindingTargetType);
+	}
+
 	@SuppressWarnings("unchecked")
 	private <T> Binder<T, ?, ?> getBinderInstance(String configurationName) {
 		if (!this.binderInstanceCache.containsKey(configurationName)) {
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java
new file mode 100644
index 000000000..0bae275ba
--- /dev/null
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java
@@ -0,0 +1,264 @@
+/*
+ * Copyright 2018 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.cloud.stream.binder;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.BiConsumer;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
+import org.springframework.context.Lifecycle;
+import org.springframework.core.AttributeAccessor;
+import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.integration.support.AckUtils;
+import org.springframework.integration.support.AcknowledgmentCallback;
+import org.springframework.integration.support.DefaultErrorMessageStrategy;
+import org.springframework.integration.support.ErrorMessageStrategy;
+import org.springframework.integration.support.ErrorMessageUtils;
+import org.springframework.integration.support.StaticMessageHeaderAccessor;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.retry.RecoveryCallback;
+import org.springframework.retry.RetryCallback;
+import org.springframework.retry.RetryContext;
+import org.springframework.retry.RetryListener;
+import org.springframework.retry.support.RetryTemplate;
+import org.springframework.util.Assert;
+
+/**
+ * The default implementation of a {@link PollableMessageSource}.
+ *
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public class DefaultPollableMessageSource implements PollableMessageSource, Lifecycle, RetryListener {
+
+	protected static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<AttributeAccessor>();
+
+	private final List<ChannelInterceptor> interceptors = new ArrayList<>();
+
+	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
+
+	private MessageSource<?> source;
+
+	private RetryTemplate retryTemplate;
+
+	private RecoveryCallback<Object> recoveryCallback;
+
+	private MessageChannel errorChannel;
+
+	private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
+
+	private BiConsumer<AttributeAccessor, Message<?>> attributesProvider;
+
+	private volatile boolean running;
+
+	public void setSource(MessageSource<?> source) {
+		ProxyFactory pf = new ProxyFactory(source);
+		class ReceiveAdvice implements MethodInterceptor {
+
+			private final List<ChannelInterceptor> interceptors = new ArrayList<>();
+
+			@Override
+			public Object invoke(MethodInvocation invocation) throws Throwable {
+				Object result = invocation.proceed();
+				if (result instanceof Message) {
+					Message<?> received = (Message<?>) result;
+					for (ChannelInterceptor interceptor : this.interceptors) {
+						received = interceptor.preSend(received, null);
+						if (received == null) {
+							return null;
+						}
+					}
+					return received;
+				}
+				return result;
+			}
+
+		}
+		final ReceiveAdvice advice = new ReceiveAdvice();
+		advice.interceptors.addAll(this.interceptors);
+		NameMatchMethodPointcutAdvisor sourceAdvisor = new NameMatchMethodPointcutAdvisor(advice);
+		sourceAdvisor.addMethodName("receive");
+		pf.addAdvisor(sourceAdvisor);
+		this.source = (MessageSource<?>) pf.getProxy();
+	}
+
+	public void setRetryTemplate(RetryTemplate retryTemplate) {
+		retryTemplate.registerListener(this);
+		this.retryTemplate = retryTemplate;
+	}
+
+	public void setRecoveryCallback(RecoveryCallback<Object> recoveryCallback) {
+		this.recoveryCallback = recoveryCallback;
+	}
+
+	public void setErrorChannel(MessageChannel errorChannel) {
+		this.errorChannel = errorChannel;
+	}
+
+	public void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) {
+		Assert.notNull(errorMessageStrategy, "'errorMessageStrategy' cannot be null");
+		this.errorMessageStrategy = errorMessageStrategy;
+	}
+
+	public void setAttributesProvider(BiConsumer<AttributeAccessor, Message<?>> attributesProvider) {
+		this.attributesProvider = attributesProvider;
+	}
+
+	public void addInterceptor(ChannelInterceptor interceptor) {
+		this.interceptors.add(interceptor);
+	}
+
+	public void addInterceptor(int index, ChannelInterceptor interceptor) {
+		this.interceptors.add(index, interceptor);
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	@Override
+	public void start() {
+		if (!this.running && this.source instanceof Lifecycle) {
+			((Lifecycle) this.source).start();
+		}
+		this.running = true;
+	}
+
+	@Override
+	public void stop() {
+		if (this.running && this.source instanceof Lifecycle) {
+			((Lifecycle) this.source).stop();
+		}
+		this.running = false;
+	}
+
+	/**
+	 * If there's a retry template, it will set the attributes holder via the listener. If
+	 * there's no retry template, but there's an error channel, we create a new attributes
+	 * holder here. If an attributes holder exists (by either method), we set the
+	 * attributes for use by the {@link ErrorMessageStrategy}.
+	 * @param message the Spring Messaging message to use.
+	 */
+	private void setAttributesIfNecessary(Message<?> message) {
+		boolean needHolder = this.errorChannel != null && this.retryTemplate == null;
+		boolean needAttributes = needHolder || this.retryTemplate != null;
+		if (needHolder) {
+			attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
+		}
+		if (needAttributes) {
+			AttributeAccessor attributes = attributesHolder.get();
+			if (attributes != null) {
+				attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
+				if (this.attributesProvider != null) {
+					this.attributesProvider.accept(attributes, message);
+				}
+			}
+		}
+	}
+
+	@Override
+	public boolean poll(MessageHandler handler) {
+		Message<?> message = this.source.receive();
+		if (message == null) {
+			return false;
+		}
+		AcknowledgmentCallback ackCallback = StaticMessageHeaderAccessor
+				.getAcknowledgmentCallback(message);
+		try {
+			if (this.retryTemplate == null && this.errorChannel == null) {
+				setAttributesIfNecessary(message);
+				doHandleMessage(handler, message);
+			}
+			else if (this.retryTemplate == null) {
+				try {
+					setAttributesIfNecessary(message);
+					doHandleMessage(handler, message);
+				}
+				catch (Exception e) {
+					if (this.errorChannel != null) {
+						this.messagingTemplate.send(this.errorChannel,
+								this.errorMessageStrategy.buildErrorMessage(e, attributesHolder.get()));
+					}
+					else {
+						throw e;
+					}
+				}
+			}
+			else {
+				this.retryTemplate.execute(context -> {
+					setAttributesIfNecessary(message);
+					doHandleMessage(handler, message);
+					return null;
+				}, this.recoveryCallback);
+			}
+			return true;
+		}
+		catch (Exception e) {
+			AckUtils.autoNack(ackCallback);
+			if (e instanceof MessageHandlingException
+					&& ((MessageHandlingException) e).getFailedMessage().equals(message)) {
+				throw (MessageHandlingException) e;
+			}
+			throw new MessageHandlingException(message, e);
+		}
+		finally {
+			AckUtils.autoAck(ackCallback);
+		}
+	}
+
+	private void doHandleMessage(MessageHandler handler, Message<?> message) {
+		try {
+			handler.handleMessage(message);
+		}
+		catch (Throwable t) { // NOSONAR
+			throw new MessageHandlingException(message, t);
+		}
+	}
+
+	@Override
+	public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
+		if (DefaultPollableMessageSource.this.recoveryCallback != null) {
+			attributesHolder.set(context);
+		}
+		return true;
+	}
+
+	@Override
+	public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
+			Throwable throwable) {
+		attributesHolder.remove();
+	}
+
+	@Override
+	public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
+			Throwable throwable) {
+		// Empty
+	}
+
+}
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableConsumerBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableConsumerBinder.java
new file mode 100644
index 000000000..e2cb3a6ec
--- /dev/null
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableConsumerBinder.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2018 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.cloud.stream.binder;
+
+/**
+ * A binder that supports pollable message sources.
+ *
+ * @param <H> the polled consumer handler type.
+ * @param <C> the consumer properties type.
+ * @param <P> the producer properties type.
+ *
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public interface PollableConsumerBinder<H, C extends ConsumerProperties> {
+
+	/**
+	 * Configure a binding for a pollable message source.
+	 * @param name the binding name.
+	 * @param group the consumer group.
+	 * @param inboundBindTarget the binding target.
+	 * @param consumerProperties the consumer properties.
+	 * @return the binding.
+	 */
+	default Binding<PollableSource<H>> bindPollableConsumer(String name, String group,
+			PollableSource<H> inboundBindTarget, C consumerProperties) {
+		throw new UnsupportedOperationException("This binder does not support pollable consumers");
+	}
+
+}
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableMessageSource.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableMessageSource.java
new file mode 100644
index 000000000..9e765ffa9
--- /dev/null
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableMessageSource.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018 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.cloud.stream.binder;
+
+import org.springframework.messaging.MessageHandler;
+
+/**
+ * A {@link PollableSource} that calls a {@link MessageHandler} with a
+ * {@link org.springframework.messaging.Message}.
+ *
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public interface PollableMessageSource extends PollableSource<MessageHandler> {
+
+}
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java
new file mode 100644
index 000000000..db3226e67
--- /dev/null
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2018 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.cloud.stream.binder;
+
+/**
+ * A mechanism to poll a consumer.
+ *
+ * @param <H> the handler type to process the result of the poll.
+ *
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public interface PollableSource<H> {
+
+	/**
+	 * Poll the consumer.
+	 * @param handler the handler.
+	 * @return true if a message was handled.
+	 */
+	boolean poll(H handler);
+
+}
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
index 02caeff12..b5c18f0af 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -35,6 +35,8 @@ import org.springframework.cloud.stream.binder.ConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
+import org.springframework.cloud.stream.binder.PollableConsumerBinder;
+import org.springframework.cloud.stream.binder.PollableSource;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.scheduling.TaskScheduler;
@@ -106,7 +108,13 @@ public class BindingService {
 		}
 		validate(consumerProperties);
 		for (String target : bindingTargets) {
-			Binding<T> binding = doBindConsumer(input, inputName, binder, consumerProperties, target);
+			Binding<T> binding;
+			if (input instanceof PollableSource) {
+				binding = doBindPollableConsumer(input, inputName, binder, consumerProperties, target);
+			}
+			else {
+				binding = doBindConsumer(input, inputName, binder, consumerProperties, target);
+			}
 			bindings.add(binding);
 		}
 		bindings = Collections.unmodifiableCollection(bindings);
@@ -152,6 +160,47 @@ public class BindingService {
 		});
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	public <T> Binding<T> doBindPollableConsumer(T input, String inputName, Binder<T, ConsumerProperties, ?> binder,
+			ConsumerProperties consumerProperties, String target) {
+		if (this.taskScheduler == null || this.bindingServiceProperties.getBindingRetryInterval() <= 0) {
+			return ((PollableConsumerBinder) binder).bindPollableConsumer(target,
+					this.bindingServiceProperties.getGroup(inputName), (PollableSource) input,
+					consumerProperties);
+		}
+		else {
+			try {
+				return ((PollableConsumerBinder) binder).bindPollableConsumer(target,
+						this.bindingServiceProperties.getGroup(inputName), (PollableSource) input,
+						consumerProperties);
+			}
+			catch (RuntimeException e) {
+				LateBinding<T> late = new LateBinding<T>();
+				reschedulePollableConsumerBinding(input, inputName, binder, consumerProperties, target, late, e);
+				return late;
+			}
+		}
+	}
+
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	public <T> void reschedulePollableConsumerBinding(final T input, final String inputName,
+			final Binder<T, ConsumerProperties, ?> binder, final ConsumerProperties consumerProperties,
+			final String target, final LateBinding<T> late, RuntimeException exception) {
+		assertNotIllegalException(exception);
+		this.log.error("Failed to create consumer binding; retrying in " +
+			this.bindingServiceProperties.getBindingRetryInterval() + " seconds", exception);
+		this.scheduleTask(() -> {
+			try {
+				late.setDelegate(((PollableConsumerBinder) binder).bindPollableConsumer(target,
+						this.bindingServiceProperties.getGroup(inputName), (PollableSource) input,
+						consumerProperties));
+			}
+			catch (RuntimeException e) {
+				reschedulePollableConsumerBinding(input, inputName, binder, consumerProperties, target, late, e);
+			}
+		});
+	}
+
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	public <T> Binding<T> bindProducer(T output, String outputName) {
 		String bindingTarget = this.bindingServiceProperties
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java
index fd235af99..f39177d13 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/CompositeMessageChannelConfigurer.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -18,13 +18,14 @@ package org.springframework.cloud.stream.binding;
 
 import java.util.List;
 
+import org.springframework.cloud.stream.binder.PollableMessageSource;
 import org.springframework.messaging.MessageChannel;
 
 /**
  * {@link MessageChannelConfigurer} that composes all the message channel configurers.
  * @author Ilayaperumal Gopinathan
  */
-public class CompositeMessageChannelConfigurer implements MessageChannelConfigurer {
+public class CompositeMessageChannelConfigurer implements MessageChannelAndSourceConfigurer {
 
 	private final List<MessageChannelConfigurer> messageChannelConfigurers;
 
@@ -46,4 +47,14 @@ public class CompositeMessageChannelConfigurer implements MessageChannelConfigur
 		}
 	}
 
+	@Override
+	public void configurePolledMessageSource(PollableMessageSource binding,
+			String name) {
+		this.messageChannelConfigurers.forEach(cconfigurer -> {
+			if (cconfigurer instanceof MessageChannelAndSourceConfigurer) {
+				((MessageChannelAndSourceConfigurer) cconfigurer).configurePolledMessageSource(binding, name);
+			}
+		});
+	}
+
 }
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageChannelAndSourceConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageChannelAndSourceConfigurer.java
new file mode 100644
index 000000000..7a9739940
--- /dev/null
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageChannelAndSourceConfigurer.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2018 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.cloud.stream.binding;
+
+import org.springframework.cloud.stream.binder.PollableMessageSource;
+
+/**
+ * Configurer for {@link PollableMessageSource}.
+ *
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public interface MessageChannelAndSourceConfigurer extends MessageChannelConfigurer {
+
+	/**
+	 * Configure the provided message source binding.
+	 * @param binding the binding.
+	 * @param name the name.
+	 */
+	void configurePolledMessageSource(PollableMessageSource binding, String name);
+
+}
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
index e3d7601c6..5e61a07e7 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -26,11 +26,13 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.cloud.stream.binder.BinderException;
 import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.DefaultPollableMessageSource;
 import org.springframework.cloud.stream.binder.JavaClassMimeTypeUtils;
 import org.springframework.cloud.stream.binder.MessageValues;
 import org.springframework.cloud.stream.binder.PartitionHandler;
 import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy;
 import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
+import org.springframework.cloud.stream.binder.PollableMessageSource;
 import org.springframework.cloud.stream.binder.ProducerProperties;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
@@ -72,7 +74,7 @@ import org.springframework.util.StringUtils;
  * @author Oleg Zhurakousky
  */
 public class MessageConverterConfigurer
-		implements MessageChannelConfigurer, BeanFactoryAware, InitializingBean {
+		implements MessageChannelAndSourceConfigurer, BeanFactoryAware, InitializingBean {
 
 	private final MessageBuilderFactory messageBuilderFactory = new MutableMessageBuilderFactory();
 
@@ -111,6 +113,18 @@ public class MessageConverterConfigurer
 		configureMessageChannel(messageChannel, channelName, false);
 	}
 
+	@Override
+	public void configurePolledMessageSource(PollableMessageSource binding, String name) {
+		BindingProperties bindingProperties = this.bindingServiceProperties.getBindingProperties(name);
+		String contentType = bindingProperties.getContentType();
+		ConsumerProperties consumerProperties = bindingProperties.getConsumer();
+		if ((consumerProperties == null || !consumerProperties.isUseNativeDecoding())
+				&& binding instanceof DefaultPollableMessageSource) {
+			((DefaultPollableMessageSource) binding).addInterceptor(
+					new InboundContentTypeConvertingInterceptor(contentType, this.compositeMessageConverterFactory));
+		}
+	}
+
 	/**
 	 * Setup data-type and message converters for the given message channel.
 	 *
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageSourceBindingTargetFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageSourceBindingTargetFactory.java
new file mode 100644
index 000000000..0f01e3977
--- /dev/null
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageSourceBindingTargetFactory.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2018 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.cloud.stream.binding;
+
+import org.springframework.cloud.stream.binder.DefaultPollableMessageSource;
+import org.springframework.cloud.stream.binder.PollableMessageSource;
+import org.springframework.util.Assert;
+
+/**
+ * An implementation of {@link BindingTargetFactory} for creating
+ * {@link DefaultPollableMessageSource}s.
+ *
+ * @author Gary Russell
+ */
+public class MessageSourceBindingTargetFactory
+		extends AbstractBindingTargetFactory<PollableMessageSource> {
+
+	private final MessageChannelAndSourceConfigurer messageSourceConfigurer;
+
+	public MessageSourceBindingTargetFactory(MessageChannelConfigurer messageSourceConfigurer) {
+		super(PollableMessageSource.class);
+		Assert.isInstanceOf(MessageChannelAndSourceConfigurer.class, messageSourceConfigurer);
+		this.messageSourceConfigurer = (MessageChannelAndSourceConfigurer) messageSourceConfigurer;
+	}
+
+	@Override
+	public PollableMessageSource createInput(String name) {
+		DefaultPollableMessageSource binding = new DefaultPollableMessageSource();
+		this.messageSourceConfigurer.configurePolledMessageSource(binding, name);
+		return binding;
+	}
+
+	@Override
+	public PollableMessageSource createOutput(String name) {
+		throw new UnsupportedOperationException();
+	}
+
+}
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
index b98cb11e2..dfd0db0d5 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
@@ -39,6 +39,7 @@ import org.springframework.cloud.stream.binding.InputBindingLifecycle;
 import org.springframework.cloud.stream.binding.MessageChannelConfigurer;
 import org.springframework.cloud.stream.binding.MessageChannelStreamListenerResultAdapter;
 import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
+import org.springframework.cloud.stream.binding.MessageSourceBindingTargetFactory;
 import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
 import org.springframework.cloud.stream.binding.SingleBindingTargetBindable;
 import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor;
@@ -132,6 +133,12 @@ public class BindingServiceConfiguration {
 		return new SubscribableChannelBindingTargetFactory(compositeMessageChannelConfigurer);
 	}
 
+	@Bean
+	public MessageSourceBindingTargetFactory messageSourceFactory(
+			CompositeMessageChannelConfigurer compositeMessageChannelConfigurer) {
+		return new MessageSourceBindingTargetFactory(compositeMessageChannelConfigurer);
+	}
+
 	@Bean
 	@ConditionalOnMissingBean
 	public CompositeMessageChannelConfigurer compositeMessageChannelConfigurer(
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/reflection/GenericsUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/reflection/GenericsUtils.java
index 0279a1e64..5590204e0 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/reflection/GenericsUtils.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/reflection/GenericsUtils.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -16,20 +16,31 @@
 
 package org.springframework.cloud.stream.reflection;
 
+import java.lang.reflect.Type;
+
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.cloud.stream.binder.PollableConsumerBinder;
+import org.springframework.cloud.stream.binder.PollableSource;
 import org.springframework.core.ResolvableType;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 
 /**
  * Internal utilities for handling generics.
  *
  * @author Marius Bogoevici
+ * @author Gary Russell
  */
-public abstract class GenericsUtils {
+public final class GenericsUtils {
+
+	private GenericsUtils() {
+		super();
+	}
 
 	/**
-	 * For a specific class that implements or extends a parametrized type returns the
+	 * For a specific class that implements or extends a parameterized type, return the
 	 * parameter of that interface at a given position. For example, for this class:
-	 * 
+	 *
 	 * <pre>
 	 * {@code
 	 * class MessageChannelBinder implements Binder<MessageChannel, ?, ?>
@@ -86,4 +97,48 @@ public abstract class GenericsUtils {
 		}
 		return bindableType;
 	}
+
+	/**
+	 * Return the generic type of PollableSource to determine if it is appropriate
+	 * for the binder.
+	 * e.g., with PollableMessageSource extends PollableSource<MessageHandler>
+	 * and  AbstractMessageChannelBinder
+	 *             implements PollableConsumerBinder<MessageHandler, C>
+	 * We're checking that the the generic type (MessageHandler) matches.
+	 *
+	 * @param binderInstance the binder.
+	 * @param bindingTargetType the binding target type.
+	 * @return
+	 */
+	@SuppressWarnings("rawtypes")
+	public static boolean checkCompatiblePollableBinder(Binder binderInstance, Class<?> bindingTargetType) {
+		Class<?>[] binderInterfaces = ClassUtils.getAllInterfaces(binderInstance);
+		for (Class<?> intf : binderInterfaces) {
+			if (PollableConsumerBinder.class.isAssignableFrom(intf)) {
+				Class<?>[] targetInterfaces = ClassUtils.getAllInterfacesForClass(bindingTargetType);
+				Class<?> psType = findPollableSourceType(targetInterfaces);
+				if (psType != null) {
+					return getParameterType(binderInstance.getClass(), intf, 0)
+							.isAssignableFrom(psType);
+				}
+			}
+		}
+		return false;
+	}
+
+	private static Class<?> findPollableSourceType(Class<?>[] targetInterfaces) {
+		for (Class<?> targetIntf : targetInterfaces) {
+			if (PollableSource.class.isAssignableFrom(targetIntf)) {
+				Type[] supers = targetIntf.getGenericInterfaces();
+				for (Type type : supers) {
+					ResolvableType resolvableType = ResolvableType.forType(type);
+					if (resolvableType.getRawClass().equals(PollableSource.class)) {
+						return resolvableType.getGeneric(0).getRawClass();
+					}
+				}
+			}
+		}
+		return null;
+	}
+
 }
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java
index f1f08a902..c6e466c35 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/AggregationTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -37,6 +37,7 @@ import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.annotation.Output;
 import org.springframework.cloud.stream.binding.BindableProxyFactory;
 import org.springframework.cloud.stream.binding.BindingTargetFactory;
+import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory;
 import org.springframework.cloud.stream.messaging.Processor;
 import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
@@ -58,7 +59,7 @@ import static org.junit.Assert.assertTrue;
  * @author Ilayaperumal Gopinathan
  * @author Artem Bilan
  * @author Janne Valkealahti
- * @author Artem Bilan
+ * @author Gary Russell
  */
 public class AggregationTest {
 
@@ -90,7 +91,7 @@ public class AggregationTest {
 		SharedBindingTargetRegistry sharedBindingTargetRegistry = aggregatedApplicationContext
 				.getBean(SharedBindingTargetRegistry.class);
 		BindingTargetFactory channelFactory = aggregatedApplicationContext
-				.getBean(BindingTargetFactory.class);
+				.getBean(SubscribableChannelBindingTargetFactory.class);
 		assertThat(channelFactory).isNotNull();
 		assertThat(sharedBindingTargetRegistry.getAll().keySet()).hasSize(2);
 		aggregatedApplicationContext.close();
@@ -102,8 +103,10 @@ public class AggregationTest {
 		aggregatedApplicationContext = new AggregateApplicationBuilder(
 				MockBinderRegistryConfiguration.class, "--server.port=0")
 						.from(TestSource.class).to(TestProcessor.class).run();
-		SharedBindingTargetRegistry sharedChannelRegistry = aggregatedApplicationContext.getBean(SharedBindingTargetRegistry.class);
-		BindingTargetFactory channelFactory = aggregatedApplicationContext.getBean(BindingTargetFactory.class);
+		SharedBindingTargetRegistry sharedChannelRegistry = aggregatedApplicationContext
+				.getBean(SharedBindingTargetRegistry.class);
+		BindingTargetFactory channelFactory = aggregatedApplicationContext
+				.getBean(SubscribableChannelBindingTargetFactory.class);
 		assertThat(channelFactory).isNotNull();
 		assertThat(sharedChannelRegistry.getAll().keySet()).hasSize(2);
 		aggregatedApplicationContext.close();
@@ -348,8 +351,10 @@ public class AggregationTest {
 				MockBinderRegistryConfiguration.class, "--server.port=0")
 						.from(TestSource.class).namespace("foo").to(TestProcessor.class)
 						.namespace("bar").run();
-		SharedBindingTargetRegistry sharedChannelRegistry = aggregatedApplicationContext.getBean(SharedBindingTargetRegistry.class);
-		BindingTargetFactory channelFactory = aggregatedApplicationContext.getBean(BindingTargetFactory.class);
+		SharedBindingTargetRegistry sharedChannelRegistry = aggregatedApplicationContext
+				.getBean(SharedBindingTargetRegistry.class);
+		BindingTargetFactory channelFactory = aggregatedApplicationContext
+				.getBean(SubscribableChannelBindingTargetFactory.class);
 		MessageChannel fooOutput = sharedChannelRegistry.get("foo.output", MessageChannel.class);
 		assertThat(fooOutput).isNotNull();
 		Object barInput = sharedChannelRegistry.get("bar.input", MessageChannel.class);
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java
index 99e4f308c..2fb09db7c 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2017 the original author or authors.
+ * Copyright 2013-2018 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.
@@ -30,11 +30,11 @@ import org.mockito.Mockito;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.stream.binder.integration.SpringIntegrationBinderConfiguration;
-import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory;
 import org.springframework.cloud.stream.binding.Bindable;
 import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
 import org.springframework.cloud.stream.binding.BindingService;
 import org.springframework.cloud.stream.binding.DynamicDestinationsBindable;
+import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.context.ConfigurableApplicationContext;
@@ -70,19 +70,20 @@ public class BinderAwareChannelResolverTests {
 
 	protected volatile Binder<MessageChannel, ConsumerProperties, ProducerProperties> binder;
 
-	protected volatile AbstractBindingTargetFactory<? extends MessageChannel> bindingTargetFactory;
+	protected volatile SubscribableChannelBindingTargetFactory bindingTargetFactory;
 
 	protected volatile BindingServiceProperties bindingServiceProperties;
 
 	protected volatile DynamicDestinationsBindable dynamicDestinationsBindable;
 
+	@SuppressWarnings("unchecked")
 	@Before
 	public void setupContext() throws Exception {
 		this.context = new SpringApplicationBuilder(SpringIntegrationBinderConfiguration.getCompleteConfiguration()).web(WebApplicationType.NONE).run();
 		this.resolver = context.getBean(BinderAwareChannelResolver.class);
 		this.binder = context.getBean(Binder.class);
 		this.bindingServiceProperties = context.getBean(BindingServiceProperties.class);
-		this.bindingTargetFactory = context.getBean(AbstractBindingTargetFactory.class);
+		this.bindingTargetFactory = context.getBean(SubscribableChannelBindingTargetFactory.class);
 	}
 
 	@Test
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java
new file mode 100644
index 000000000..aa0ddfc4d
--- /dev/null
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java
@@ -0,0 +1,185 @@
+/*
+ * Copyright 2018 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.cloud.stream.binder;
+
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import org.springframework.cloud.stream.binder.integration.SpringIntegrationChannelBinder;
+import org.springframework.cloud.stream.binder.integration.SpringIntegrationProvisioner;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.integration.channel.PublishSubscribeChannel;
+import org.springframework.integration.context.IntegrationContextUtils;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.ChannelInterceptorAdapter;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.messaging.support.MessageBuilder;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+
+/**
+ * @author Gary Russell
+ * @since 2.0
+ *
+ */
+public class PollableConsumerTests {
+
+	private final GenericApplicationContext context = new GenericApplicationContext();
+
+	@Test
+	public void testSimple() {
+		SpringIntegrationChannelBinder binder = createBinder();
+		DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource();
+		pollableSource.addInterceptor(new ChannelInterceptorAdapter() {
+
+			@Override
+			public Message<?> preSend(Message<?> message, MessageChannel channel) {
+				return MessageBuilder.withPayload(((String) message.getPayload()).toUpperCase())
+						.copyHeaders(message.getHeaders())
+						.build();
+			}
+
+		});
+		ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(null);
+		properties.setMaxAttempts(2);
+		properties.setBackOffInitialInterval(0);
+		binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
+		final AtomicInteger count = new AtomicInteger();
+		assertThat(pollableSource.poll(received -> {
+			assertThat(received.getPayload()).isEqualTo("POLLED DATA");
+			assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo("text/plain");
+			if (count.incrementAndGet() == 1) {
+				throw new RuntimeException("test retry");
+			}
+		})).isTrue();
+		assertThat(count.get()).isEqualTo(2);
+	}
+
+	@Test
+	public void testEmbedded() {
+		SpringIntegrationChannelBinder binder = createBinder();
+		binder.setMessageSourceDelegate(() -> {
+			MessageValues original = new MessageValues("foo".getBytes(),
+					Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "application/octet-stream"));
+			byte[] payload = new byte[0];
+			try {
+				payload = EmbeddedHeaderUtils.embedHeaders(original, MessageHeaders.CONTENT_TYPE);
+			}
+			catch (Exception e) {
+				fail(e.getMessage());
+			}
+			return new GenericMessage<>(payload);
+		});
+		ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(null);
+		properties.setHeaderMode(HeaderMode.embeddedHeaders);
+		DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource();
+		pollableSource.addInterceptor(new ChannelInterceptorAdapter() {
+
+			@Override
+			public Message<?> preSend(Message<?> message, MessageChannel channel) {
+				return MessageBuilder.withPayload(new String((byte[]) message.getPayload()).toUpperCase())
+						.copyHeaders(message.getHeaders())
+						.build();
+			}
+
+		});
+		binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
+		assertThat(pollableSource.poll(received -> {
+			assertThat(received.getPayload()).isEqualTo("FOO");
+			assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE)).isEqualTo("application/octet-stream");
+		})).isTrue();
+	}
+
+	@Test
+	public void testErrors() {
+		SpringIntegrationChannelBinder binder = createBinder();
+		DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource();
+		pollableSource.addInterceptor(new ChannelInterceptorAdapter() {
+
+			@Override
+			public Message<?> preSend(Message<?> message, MessageChannel channel) {
+				return MessageBuilder.withPayload(((String) message.getPayload()).toUpperCase())
+						.copyHeaders(message.getHeaders())
+						.build();
+			}
+
+		});
+		ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(null);
+		properties.setMaxAttempts(2);
+		properties.setBackOffInitialInterval(0);
+		binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
+		final CountDownLatch latch = new CountDownLatch(1);
+		this.context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class).subscribe(m -> {
+			latch.countDown();
+		});
+		final AtomicInteger count = new AtomicInteger();
+		assertThat(pollableSource.poll(received -> {
+			count.incrementAndGet();
+			throw new RuntimeException("test recoverer");
+		})).isTrue();
+		assertThat(count.get()).isEqualTo(2);
+		Message<?> lastError = binder.getLastError();
+		assertThat(lastError).isNotNull();
+		assertThat(((Exception) lastError.getPayload()).getCause().getMessage()).isEqualTo("test recoverer");
+	}
+
+	@Test
+	public void testErrorsNoRetry() {
+		SpringIntegrationChannelBinder binder = createBinder();
+		DefaultPollableMessageSource pollableSource = new DefaultPollableMessageSource();
+		pollableSource.addInterceptor(new ChannelInterceptorAdapter() {
+
+			@Override
+			public Message<?> preSend(Message<?> message, MessageChannel channel) {
+				return MessageBuilder.withPayload(((String) message.getPayload()).toUpperCase())
+						.copyHeaders(message.getHeaders())
+						.build();
+			}
+
+		});
+		ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(null);
+		properties.setMaxAttempts(1);
+		binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
+		final CountDownLatch latch = new CountDownLatch(1);
+		this.context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class).subscribe(m -> {
+			latch.countDown();
+		});
+		final AtomicInteger count = new AtomicInteger();
+		assertThat(pollableSource.poll(received -> {
+			count.incrementAndGet();
+			throw new RuntimeException("test recoverer");
+		})).isTrue();
+		assertThat(count.get()).isEqualTo(1);
+	}
+
+	private SpringIntegrationChannelBinder createBinder() {
+		SpringIntegrationProvisioner provisioningProvider = new SpringIntegrationProvisioner();
+		SpringIntegrationChannelBinder binder = new SpringIntegrationChannelBinder(provisioningProvider);
+		this.context.registerBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, PublishSubscribeChannel.class);
+		this.context.refresh();
+		binder.setApplicationContext(this.context);
+		return binder;
+	}
+
+}
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SampleStreamApp.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SampleStreamApp.java
index 23bed3c72..4228c54e0 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SampleStreamApp.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SampleStreamApp.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.
@@ -18,13 +18,17 @@ package org.springframework.cloud.stream.binder.integration;
 
 import java.nio.charset.StandardCharsets;
 
+import org.springframework.boot.ApplicationRunner;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.Input;
 import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.binder.PollableMessageSource;
 import org.springframework.cloud.stream.messaging.Processor;
 import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.messaging.Message;
@@ -37,10 +41,11 @@ import static org.junit.Assert.assertEquals;
  * Sample spring cloud stream application that demonstrates the usage of {@link SpringIntegrationChannelBinder}.
  *
  * @author Oleg Zhurakousky
+ * @author Gary Russell
  *
  */
 @SpringBootApplication
-@EnableBinding(Processor.class)
+@EnableBinding(SampleStreamApp.PolledConsumer.class)
 @Import(SpringIntegrationBinderConfiguration.class)
 public class SampleStreamApp {
 
@@ -55,6 +60,13 @@ public class SampleStreamApp {
 		assertEquals("Hello", new String((byte[])message.getPayload(), StandardCharsets.UTF_8));
 	}
 
+	@Bean
+	public ApplicationRunner runner(PollableMessageSource pollableSource) {
+		return args -> pollableSource.poll(message -> {
+			System.out.println("Polled payload: " + message.getPayload());
+		});
+	}
+
 	@StreamListener(Processor.INPUT)
 	@SendTo(Processor.OUTPUT)
 	public String receive(String value) {
@@ -66,6 +78,14 @@ public class SampleStreamApp {
 	public void error(String value) {
 		System.out.println("Handling ERROR payload: " + value);
 	}
+
+	public interface PolledConsumer extends Processor {
+
+		@Input("pollableSource")
+		PollableMessageSource pollableSource();
+
+	}
+
 }
 
 
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java
index 3f5d565b4..b1a0248e6 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationChannelBinder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.stream.binder.integration;
 
+import java.util.Collections;
 import java.util.function.Consumer;
 
 import org.springframework.beans.factory.BeanFactory;
@@ -30,6 +31,7 @@ import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.core.AttributeAccessor;
 import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.handler.BridgeHandler;
 import org.springframework.integration.support.DefaultErrorMessageStrategy;
@@ -37,8 +39,10 @@ import org.springframework.integration.support.ErrorMessageStrategy;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.RetryCallback;
 import org.springframework.retry.RetryContext;
@@ -94,17 +98,40 @@ import org.springframework.util.StringUtils;
  * </pre>
  *
  * @author Oleg Zhurakousky
+ * @author Gary Russell
+ *
  */
-class SpringIntegrationChannelBinder extends AbstractMessageChannelBinder<ConsumerProperties,
+public class SpringIntegrationChannelBinder extends AbstractMessageChannelBinder<ConsumerProperties,
 	ProducerProperties, SpringIntegrationProvisioner> {
 
 	@Autowired
 	private BeanFactory beanFactory;
 
-	SpringIntegrationChannelBinder(SpringIntegrationProvisioner provisioningProvider) {
+	private Message<?> lastError;
+
+	private MessageSource<?> messageSourceDelegate = () -> new GenericMessage<>("polled data",
+			Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "text/plain"));
+
+	public SpringIntegrationChannelBinder(SpringIntegrationProvisioner provisioningProvider) {
 		super(new String[] {}, provisioningProvider);
 	}
 
+	/**
+	 * Set a delegate {@link MessageSource} for pollable consumers.
+	 * @param messageSourceDelegate the delegate.
+	 */
+	public void setMessageSourceDelegate(MessageSource<?> messageSourceDelegate) {
+		this.messageSourceDelegate = messageSourceDelegate;
+	}
+
+	public Message<?> getLastError() {
+		return this.lastError;
+	}
+
+	public void setLastError(Message<?> lastError) {
+		this.lastError = lastError;
+	}
+
 	@Override
 	protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
 			ProducerProperties producerProperties, MessageChannel errorChannel) throws Exception {
@@ -139,6 +166,22 @@ class SpringIntegrationChannelBinder extends AbstractMessageChannelBinder<Consum
 		return adapter;
 	}
 
+	@Override
+	protected PolledConsumerResources createPolledConsumerResources(String name, String group, ConsumerDestination destination,
+			ConsumerProperties consumerProperties) {
+		return new PolledConsumerResources(this.messageSourceDelegate,
+				registerErrorInfrastructure(destination, group, consumerProperties));
+	}
+
+	@Override
+	protected MessageHandler getErrorMessageHandler(ConsumerDestination destination, String group,
+			ConsumerProperties consumerProperties) {
+		return m -> {
+			this.logger.debug("Error handled: " + m);
+			this.lastError = m;
+		};
+	}
+
 	/**
 	 * Implementation of simple message listener container modeled after AMQP SimpleMessageListenerContainer
 	 */
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationProvisioner.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationProvisioner.java
index ddd3a8f46..35d111fd2 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationProvisioner.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/integration/SpringIntegrationProvisioner.java
@@ -42,7 +42,7 @@ import org.springframework.messaging.SubscribableChannel;
  * @author Oleg Zhurakousky
  *
  */
-class SpringIntegrationProvisioner implements ProvisioningProvider<ConsumerProperties, ProducerProperties> {
+public class SpringIntegrationProvisioner implements ProvisioningProvider<ConsumerProperties, ProducerProperties> {
 
 	private final Map<String, SubscribableChannel> provisionedDestinations = new HashMap<>();
 
@@ -72,7 +72,9 @@ class SpringIntegrationProvisioner implements ProvisioningProvider<ConsumerPrope
 	@Override
 	public ConsumerDestination provisionConsumerDestination(String name, String group, ConsumerProperties properties) throws ProvisioningException {
 		SubscribableChannel destination = this.provisionDestination(name, false);
-		this.source.setChannel(destination);
+		if (this.source != null) {
+			this.source.setChannel(destination);
+		}
 		return new SpringIntegrationConsumerDestination(name, destination);
 	}