From 7267661d2e1b34ad9b80409f5b2195d62d84ebad Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 10 Jan 2018 09:20:31 -0500 Subject: [PATCH] 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 --- .../AbstractPollableConsumerTestBinder.java | 45 +++ spring-cloud-stream/pom.xml | 1 + .../binder/AbstractMessageChannelBinder.java | 130 ++++++++- .../cloud/stream/binder/Binder.java | 6 +- .../stream/binder/DefaultBinderFactory.java | 21 +- .../binder/DefaultPollableMessageSource.java | 264 ++++++++++++++++++ .../stream/binder/PollableConsumerBinder.java | 45 +++ .../stream/binder/PollableMessageSource.java | 31 ++ .../cloud/stream/binder/PollableSource.java | 37 +++ .../cloud/stream/binding/BindingService.java | 53 +++- .../CompositeMessageChannelConfigurer.java | 15 +- .../MessageChannelAndSourceConfigurer.java | 37 +++ .../binding/MessageConverterConfigurer.java | 18 +- .../MessageSourceBindingTargetFactory.java | 52 ++++ .../config/BindingServiceConfiguration.java | 7 + .../stream/reflection/GenericsUtils.java | 63 ++++- .../stream/aggregation/AggregationTest.java | 19 +- .../BinderAwareChannelResolverTests.java | 9 +- .../stream/binder/PollableConsumerTests.java | 185 ++++++++++++ .../binder/integration/SampleStreamApp.java | 24 +- .../SpringIntegrationChannelBinder.java | 49 +++- .../SpringIntegrationProvisioner.java | 6 +- 22 files changed, 1081 insertions(+), 36 deletions(-) create mode 100644 spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/AbstractPollableConsumerTestBinder.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableConsumerBinder.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableMessageSource.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PollableSource.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageChannelAndSourceConfigurer.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageSourceBindingTargetFactory.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java 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, + CP extends ConsumerProperties, PP extends ProducerProperties> extends AbstractTestBinder + implements PollableConsumerBinder{ + + private PollableConsumerBinder binder; + + @SuppressWarnings("unchecked") + public void setPollableConsumerBinder(PollableConsumerBinder binder) { + super.setBinder((C) binder); + this.binder = binder; + } + + @Override + public Binding> bindPollableConsumer(String name, String group, + PollableSource 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 @@ org.springframework.integration spring-integration-core + 5.0.1.BUILD-SNAPSHOT org.springframework.integration 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> - extends AbstractBinder { + extends AbstractBinder implements PollableConsumerBinder { private final EmbeddedHeadersChannelInterceptor embeddedHeadersChannelInterceptor = new EmbeddedHeadersChannelInterceptor(this.logger); @@ -286,6 +289,65 @@ public abstract class AbstractMessageChannelBinder> bindPollableConsumer(String name, String group, + final PollableSource 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>(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 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 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 the primary binding type (e.g. MessageChannel). + * @param the consumer properties type. + * @param

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 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 boolean verifyBinderTypeMatchesTarget(Binder binderInstance, + Class bindingTargetType) { + return (binderInstance instanceof PollableConsumerBinder + && GenericsUtils.checkCompatiblePollableBinder(binderInstance, bindingTargetType)) + || GenericsUtils.getParameterType(binderInstance.getClass(), Binder.class, 0) + .isAssignableFrom(bindingTargetType); + } + @SuppressWarnings("unchecked") private Binder 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 attributesHolder = new ThreadLocal(); + + private final List interceptors = new ArrayList<>(); + + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); + + private MessageSource source; + + private RetryTemplate retryTemplate; + + private RecoveryCallback recoveryCallback; + + private MessageChannel errorChannel; + + private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); + + private BiConsumer> attributesProvider; + + private volatile boolean running; + + public void setSource(MessageSource source) { + ProxyFactory pf = new ProxyFactory(source); + class ReceiveAdvice implements MethodInterceptor { + + private final List 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 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> 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 boolean open(RetryContext context, RetryCallback callback) { + if (DefaultPollableMessageSource.this.recoveryCallback != null) { + attributesHolder.set(context); + } + return true; + } + + @Override + public void close(RetryContext context, RetryCallback callback, + Throwable throwable) { + attributesHolder.remove(); + } + + @Override + public void onError(RetryContext context, RetryCallback 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 the polled consumer handler type. + * @param the consumer properties type. + * @param

the producer properties type. + * + * @author Gary Russell + * @since 2.0 + * + */ +public interface PollableConsumerBinder { + + /** + * 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> bindPollableConsumer(String name, String group, + PollableSource 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 { + +} 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 the handler type to process the result of the poll. + * + * @author Gary Russell + * @since 2.0 + * + */ +public interface PollableSource { + + /** + * 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 binding = doBindConsumer(input, inputName, binder, consumerProperties, target); + Binding 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 Binding doBindPollableConsumer(T input, String inputName, Binder 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 late = new LateBinding(); + reschedulePollableConsumerBinding(input, inputName, binder, consumerProperties, target, late, e); + return late; + } + } + } + + @SuppressWarnings({ "rawtypes", "unchecked" }) + public void reschedulePollableConsumerBinding(final T input, final String inputName, + final Binder binder, final ConsumerProperties consumerProperties, + final String target, final LateBinding 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 Binding 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 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 { + + 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: - * + * * 

 	 * {@code
 	 * class MessageChannelBinder implements Binder
@@ -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
+	 * and  AbstractMessageChannelBinder
+	 *             implements PollableConsumerBinder
+	 * 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 binder;
 
-	protected volatile AbstractBindingTargetFactory 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 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 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 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 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;
  * 
  *
  * @author Oleg Zhurakousky
+ * @author Gary Russell
+ *
  */
-class SpringIntegrationChannelBinder extends AbstractMessageChannelBinder {
 
 	@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 {
+			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 {
+public class SpringIntegrationProvisioner implements ProvisioningProvider {
 
 	private final Map provisionedDestinations = new HashMap<>();
 
@@ -72,7 +72,9 @@ class SpringIntegrationProvisioner implements ProvisioningProvider