GH-1174: Add Pollable Consumer

Resolves: https://github.com/spring-cloud/spring-cloud-stream/issues/1174

Introduce `PollableMessageSource` for polled consumers.

The `@Input` annotation can now be applied to interface methods returning
`PollableMessageSource` and binders that can support polled consumers
will wire up an implementation `DefaultPollableMessageSource` which can
then be `poll()`ed.

The `poll()` method takes a `MessageHandler` callback to handle any message
returned by the poll; the method returns true if a message was found and
handled, false otherwise.

By default, acknowledging the message is deferred until the message handler
returns but that auto-acknowledging can be disabled if the user wishes to
defer the acknowledgment until later. Refer to the Spring Integration
documentation for more information (internally, a `MessageSourcePollingTemplate`
is used, which performs the ack/nack when the handler exits).

Usage:

```java
public interface PolledConsumer extends Processor {

	@Input("pollableSource")
	PollableMessageSource pollableSource();

}
```

and

```java
@Bean
public ApplicationRunner runner(PollableMessageSource pollableSource) {
	return args -> pollableSource.poll(message -> {
		System.out.println("Polled payload: " + message.getPayload());
	});
}
```

Polishing

Fix test; add support to test binder.

Add error channel handling and retry

More polishing driven by Rabbit implementation

- can't use the MessageSourcePollingTemplate within retry since it will fetch more messages
- populate the retry context with data for the error message strategy

Polishing - PR Comments

Fix NPE.

Revert finally in DefaultPollableMessageSource

Catch Throwable - JUnit throws Errors

Change method name to reflect it's only used for polled consumers.

Fix errors when no retry

Remove stack trace print

Fix error message strategy context with no retry

Add separate error MessageHandler for polled consumers.

Resolves #1174
Resolves #1176

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);
+	}
+
+}

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>

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;
+		}
+
+	}
+
 }

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

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)) {

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
+	}
+
+}

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");
+	}
+
+}

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> {
+
+}

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);
+
+}

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

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);
+			}
+		});
+	}
+
 }

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);
+
+}

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.
 	 *

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();
+	}
+
+}

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(

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;
+	}
+
 }

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);

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

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;
+	}
+
+}

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();
+
+	}
+
 }
 
 

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
 	 */

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);
 	}