org.springframework.cloud.stream.binder

Class AbstractMessageChannelBinder<C extends ConsumerProperties,P extends ProducerProperties,PP extends ProvisioningProvider<C,P>>

java.lang.Object

org.springframework.cloud.stream.binder.AbstractBinder<org.springframework.messaging.MessageChannel,C,P>

org.springframework.cloud.stream.binder.AbstractMessageChannelBinder<C,P,PP>

Type Parameters:

C - the consumer properties type

P - the producer properties type

All Implemented Interfaces:

org.springframework.beans.factory.Aware, org.springframework.beans.factory.InitializingBean, Binder<org.springframework.messaging.MessageChannel,C,P>, PollableConsumerBinder<org.springframework.messaging.MessageHandler,C>, org.springframework.context.ApplicationContextAware, org.springframework.context.ApplicationEventPublisherAware

Direct Known Subclasses:

KafkaMessageChannelBinder, RabbitMessageChannelBinder

public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,P extends ProducerProperties,PP extends ProvisioningProvider<C,P>>
extends AbstractBinder<org.springframework.messaging.MessageChannel,C,P>
implements PollableConsumerBinder<org.springframework.messaging.MessageHandler,C>, org.springframework.context.ApplicationEventPublisherAware

AbstractBinder that serves as base class for MessageChannel binders. Implementors must implement the following methods:

• createProducerMessageHandler(ProducerDestination, ProducerProperties, MessageChannel)
• createConsumerEndpoint(ConsumerDestination, String, ConsumerProperties)

Since:

1.1

Author:

Marius Bogoevici, Ilayaperumal Gopinathan, Soby Chacko, Oleg Zhurakousky, Artem Bilan, Gary Russell

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	AbstractMessageChannelBinder.ErrorInfrastructure
protected static class	AbstractMessageChannelBinder.PolledConsumerResources

Field Summary

Fields

Modifier and Type	Field and Description
protected PP	provisioningProvider ProvisioningProvider delegated by the downstream binder implementations.

Fields inherited from class org.springframework.cloud.stream.binder.AbstractBinder

logger

Constructor Summary

Constructors

Constructor and Description
AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider)
AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider, ListenerContainerCustomizer<?> containerCustomizer)

Method Summary

Modifier and Type	Method and Description
protected void	afterUnbindConsumer(ConsumerDestination destination, String group, C consumerProperties) Invoked after the unbinding of a consumer.
protected void	afterUnbindProducer(ProducerDestination destination, P producerProperties) Invoked after the unbinding of a producer.
Binding<PollableSource<org.springframework.messaging.MessageHandler>>	bindPollableConsumer(String name, String group, PollableSource<org.springframework.messaging.MessageHandler> inboundBindTarget, C properties) Configure a binding for a pollable message source.
protected abstract org.springframework.integration.core.MessageProducer	createConsumerEndpoint(ConsumerDestination destination, String group, C properties) Creates MessageProducer that receives data from the consumer destination.
protected AbstractMessageChannelBinder.PolledConsumerResources	createPolledConsumerResources(String name, String group, ConsumerDestination destination, C consumerProperties)
protected abstract org.springframework.messaging.MessageHandler	createProducerMessageHandler(ProducerDestination destination, P producerProperties, org.springframework.messaging.MessageChannel errorChannel) Create a MessageHandler with the ability to send data to the target middleware.
protected org.springframework.messaging.MessageHandler	createProducerMessageHandler(ProducerDestination destination, P producerProperties, org.springframework.messaging.MessageChannel channel, org.springframework.messaging.MessageChannel errorChannel) Create a MessageHandler with the ability to send data to the target middleware.
Binding<org.springframework.messaging.MessageChannel>	doBindConsumer(String name, String group, org.springframework.messaging.MessageChannel inputChannel, C properties) Binds an inbound channel to a given destination.
Binding<org.springframework.messaging.MessageChannel>	doBindProducer(String destination, org.springframework.messaging.MessageChannel outputChannel, P producerProperties) Binds an outbound channel to a given destination.
protected String	errorsBaseName(ConsumerDestination destination, String group, C consumerProperties)
protected String	errorsBaseName(ProducerDestination destination)
protected org.springframework.context.ApplicationEventPublisher	getApplicationEventPublisher()
protected <L> ListenerContainerCustomizer<L>	getContainerCustomizer()
protected org.springframework.messaging.MessageHandler	getDefaultErrorMessageHandler(org.springframework.cloud.stream.binder.LastSubscriberAwareChannel errorChannel, boolean defaultErrorChannelPresent) Return the default error message handler, which throws the error message payload to the caller if there are no user handlers subscribed.
protected String	getErrorBridgeName(ConsumerDestination destination, String group, C consumerProperties)
protected String	getErrorBridgeName(ProducerDestination destination)
protected org.springframework.messaging.MessageHandler	getErrorMessageHandler(ConsumerDestination destination, String group, C consumerProperties) Binders can return a message handler to be subscribed to the error channel.
protected String	getErrorMessageHandlerName(ConsumerDestination destination, String group, C consumerProperties)
protected org.springframework.integration.support.ErrorMessageStrategy	getErrorMessageStrategy() Binders can return an ErrorMessageStrategy for building error messages; binder implementations typically might add extra headers to the error message.
protected String	getErrorRecovererName(ConsumerDestination destination, String group, C consumerProperties)
protected org.springframework.messaging.MessageHandler	getPolledConsumerErrorMessageHandler(ConsumerDestination destination, String group, C consumerProperties) Binders can return a message handler to be subscribed to the error channel.
protected org.springframework.retry.RecoveryCallback<Object>	getPolledConsumerRecoveryCallback(AbstractMessageChannelBinder.ErrorInfrastructure errorInfrastructure, C properties) Implementations can override the default ErrorMessageSendingRecoverer.
protected void	postProcessOutputChannel(org.springframework.messaging.MessageChannel outputChannel, P producerProperties) Allows subclasses to perform post processing on the channel - for example to add more interceptors.
protected void	postProcessPollableSource(DefaultPollableMessageSource bindingTarget)
protected AbstractMessageChannelBinder.ErrorInfrastructure	registerErrorInfrastructure(ConsumerDestination destination, String group, C consumerProperties) Build an errorChannelRecoverer that writes to a pub/sub channel for the destination when an exception is thrown to a consumer.
protected AbstractMessageChannelBinder.ErrorInfrastructure	registerErrorInfrastructure(ConsumerDestination destination, String group, C consumerProperties, boolean polled) Build an errorChannelRecoverer that writes to a pub/sub channel for the destination when an exception is thrown to a consumer.
void	setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher applicationEventPublisher)
protected boolean	useNativeEncoding(P producerProperties) Whether the producer for the destination being created should be configured to use native encoding which may, or may not, be determined from the properties.

Methods inherited from class org.springframework.cloud.stream.binder.AbstractBinder

afterPropertiesSet, applyPrefix, bindConsumer, bindProducer, buildPartitionRoutingExpression, buildRetryTemplate, constructDLQName, getApplicationContext, getBeanFactory, getEvaluationContext, groupedName, onInit, serializePayloadIfNecessary, setApplicationContext

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

provisioningProvider

protected final PP extends ProvisioningProvider<C,P> provisioningProvider

ProvisioningProvider delegated by the downstream binder implementations.

Constructor Detail

AbstractMessageChannelBinder

public AbstractMessageChannelBinder(String[] headersToEmbed,
                                    PP provisioningProvider)

AbstractMessageChannelBinder

public AbstractMessageChannelBinder(String[] headersToEmbed,
                                    PP provisioningProvider,
                                    ListenerContainerCustomizer<?> containerCustomizer)

Method Detail

setApplicationEventPublisher

public void setApplicationEventPublisher(org.springframework.context.ApplicationEventPublisher applicationEventPublisher)

Specified by:

setApplicationEventPublisher in interface org.springframework.context.ApplicationEventPublisherAware

getApplicationEventPublisher

protected org.springframework.context.ApplicationEventPublisher getApplicationEventPublisher()

getContainerCustomizer

protected <L> ListenerContainerCustomizer<L> getContainerCustomizer()

doBindProducer

public final Binding<org.springframework.messaging.MessageChannel> doBindProducer(String destination,
                                                                                  org.springframework.messaging.MessageChannel outputChannel,
                                                                                  P producerProperties)
                                                                           throws BinderException

Binds an outbound channel to a given destination. The implementation delegates to ProvisioningProvider.provisionProducerDestination(String, ProducerProperties) and createProducerMessageHandler(ProducerDestination, ProducerProperties, MessageChannel) for handling the middleware specific logic. If the returned producer message handler is an InitializingBean then InitializingBean.afterPropertiesSet() will be called on it. Similarly, if the returned producer message handler endpoint is a Lifecycle, then Lifecycle.start() will be called on it.

Specified by:

doBindProducer in class AbstractBinder<org.springframework.messaging.MessageChannel,C extends ConsumerProperties,P extends ProducerProperties>

Parameters:

destination - the name of the destination

outputChannel - the channel to be bound

producerProperties - the ProducerProperties of the binding

Returns:

the Binding for the channel

Throws:

BinderException - on internal errors during binding

useNativeEncoding

protected boolean useNativeEncoding(P producerProperties)

Whether the producer for the destination being created should be configured to use native encoding which may, or may not, be determined from the properties. For example, a transactional kafka binder uses a common producer for all destinations. The default implementation returns P#isUseNativeEncoding().

Parameters:

producerProperties - the properties.

Returns:

true to use native encoding.

postProcessOutputChannel

protected void postProcessOutputChannel(org.springframework.messaging.MessageChannel outputChannel,
                                        P producerProperties)

Allows subclasses to perform post processing on the channel - for example to add more interceptors.

Parameters:

outputChannel - the channel.

producerProperties - the producer properties.

createProducerMessageHandler

protected org.springframework.messaging.MessageHandler createProducerMessageHandler(ProducerDestination destination,
                                                                                    P producerProperties,
                                                                                    org.springframework.messaging.MessageChannel channel,
                                                                                    org.springframework.messaging.MessageChannel errorChannel)
                                                                             throws Exception

Create a MessageHandler with the ability to send data to the target middleware. If the returned instance is also a Lifecycle, it will be stopped automatically by the binder.

In order to be fully compliant, the MessageHandler of the binder must observe the following headers:

• BinderHeaders.PARTITION_HEADER - indicates the target partition where the message must be sent

Parameters:

destination - the name of the target destination.

producerProperties - the producer properties.

channel - the channel to bind.

errorChannel - the error channel (if enabled, otherwise null). If not null, the binder must wire this channel into the producer endpoint so that errors are forwarded to it.

Returns:

the message handler for sending data to the target middleware

Throws:

Exception

createProducerMessageHandler

protected abstract org.springframework.messaging.MessageHandler createProducerMessageHandler(ProducerDestination destination,
                                                                                             P producerProperties,
                                                                                             org.springframework.messaging.MessageChannel errorChannel)
                                                                                      throws Exception

Create a MessageHandler with the ability to send data to the target middleware. If the returned instance is also a Lifecycle, it will be stopped automatically by the binder.

In order to be fully compliant, the MessageHandler of the binder must observe the following headers:

• BinderHeaders.PARTITION_HEADER - indicates the target partition where the message must be sent

Parameters:

destination - the name of the target destination

producerProperties - the producer properties

errorChannel - the error channel (if enabled, otherwise null). If not null, the binder must wire this channel into the producer endpoint so that errors are forwarded to it.

Returns:

the message handler for sending data to the target middleware

Throws:

Exception

afterUnbindProducer

protected void afterUnbindProducer(ProducerDestination destination,
                                   P producerProperties)

Invoked after the unbinding of a producer. Subclasses may override this to provide their own logic for dealing with unbinding.

Parameters:

destination - the bound destination

producerProperties - the producer properties

doBindConsumer

public final Binding<org.springframework.messaging.MessageChannel> doBindConsumer(String name,
                                                                                  String group,
                                                                                  org.springframework.messaging.MessageChannel inputChannel,
                                                                                  C properties)
                                                                           throws BinderException

Binds an inbound channel to a given destination. The implementation delegates to ProvisioningProvider.provisionConsumerDestination(String, String, ConsumerProperties) and createConsumerEndpoint(ConsumerDestination, String, ConsumerProperties) for handling middleware-specific logic. If the returned consumer endpoint is an InitializingBean then InitializingBean.afterPropertiesSet() will be called on it. Similarly, if the returned consumer endpoint is a Lifecycle, then Lifecycle.start() will be called on it.

Specified by:

doBindConsumer in class AbstractBinder<org.springframework.messaging.MessageChannel,C extends ConsumerProperties,P extends ProducerProperties>

Parameters:

name - the name of the destination

group - the consumer group

inputChannel - the channel to be bound

properties - the ConsumerProperties of the binding

Returns:

the Binding for the channel

Throws:

BinderException - on internal errors during binding

bindPollableConsumer

public Binding<PollableSource<org.springframework.messaging.MessageHandler>> bindPollableConsumer(String name,
                                                                                                  String group,
                                                                                                  PollableSource<org.springframework.messaging.MessageHandler> inboundBindTarget,
                                                                                                  C properties)

Description copied from interface: PollableConsumerBinder

Configure a binding for a pollable message source.

Specified by:

bindPollableConsumer in interface PollableConsumerBinder<org.springframework.messaging.MessageHandler,C extends ConsumerProperties>

Parameters:

name - the binding name.

group - the consumer group.

inboundBindTarget - the binding target.

properties - the consumer properties.

Returns:

the binding.

postProcessPollableSource

protected void postProcessPollableSource(DefaultPollableMessageSource bindingTarget)

getPolledConsumerRecoveryCallback

protected org.springframework.retry.RecoveryCallback<Object> getPolledConsumerRecoveryCallback(AbstractMessageChannelBinder.ErrorInfrastructure errorInfrastructure,
                                                                                               C properties)

Implementations can override the default ErrorMessageSendingRecoverer.

Parameters:

errorInfrastructure - the infrastructure.

properties - the consumer properties.

Returns:

the recoverer.

createPolledConsumerResources

protected AbstractMessageChannelBinder.PolledConsumerResources createPolledConsumerResources(String name,
                                                                                             String group,
                                                                                             ConsumerDestination destination,
                                                                                             C consumerProperties)

createConsumerEndpoint

protected abstract org.springframework.integration.core.MessageProducer createConsumerEndpoint(ConsumerDestination destination,
                                                                                               String group,
                                                                                               C properties)
                                                                                        throws Exception

Creates MessageProducer that receives data from the consumer destination. will be started and stopped by the binder.

Parameters:

group - the consumer group

destination - reference to the consumer destination

properties - the consumer properties

Returns:

the consumer endpoint.

Throws:

Exception

afterUnbindConsumer

protected void afterUnbindConsumer(ConsumerDestination destination,
                                   String group,
                                   C consumerProperties)

Invoked after the unbinding of a consumer. The binder implementation can override this method to provide their own logic (e.g. for cleaning up destinations).

Parameters:

destination - the consumer destination

group - the consumer group

consumerProperties - the consumer properties

registerErrorInfrastructure

protected final AbstractMessageChannelBinder.ErrorInfrastructure registerErrorInfrastructure(ConsumerDestination destination,
                                                                                             String group,
                                                                                             C consumerProperties)

Build an errorChannelRecoverer that writes to a pub/sub channel for the destination when an exception is thrown to a consumer.

Parameters:

destination - the destination.

group - the group.

consumerProperties - the properties.

Returns:

the ErrorInfrastructure which is a holder for the error channel, the recoverer and the message handler that is subscribed to the channel.

registerErrorInfrastructure

protected final AbstractMessageChannelBinder.ErrorInfrastructure registerErrorInfrastructure(ConsumerDestination destination,
                                                                                             String group,
                                                                                             C consumerProperties,
                                                                                             boolean polled)

Build an errorChannelRecoverer that writes to a pub/sub channel for the destination when an exception is thrown to a consumer.

Parameters:

destination - the destination.

group - the group.

consumerProperties - the properties.

polled - true if this is for a polled consumer.

Returns:

the ErrorInfrastructure which is a holder for the error channel, the recoverer and the message handler that is subscribed to the channel.

getErrorMessageHandler

protected org.springframework.messaging.MessageHandler getErrorMessageHandler(ConsumerDestination destination,
                                                                              String group,
                                                                              C consumerProperties)

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.

Parameters:

destination - the destination.

group - the group.

consumerProperties - the properties.

Returns:

the handler (may be null, which is the default, causing the exception to be rethrown).

getPolledConsumerErrorMessageHandler

protected org.springframework.messaging.MessageHandler getPolledConsumerErrorMessageHandler(ConsumerDestination destination,
                                                                                            String group,
                                                                                            C consumerProperties)

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.

Parameters:

destination - the destination.

group - the group.

consumerProperties - the properties.

Returns:

the handler (may be null, which is the default, causing the exception to be rethrown).

getDefaultErrorMessageHandler

protected org.springframework.messaging.MessageHandler getDefaultErrorMessageHandler(org.springframework.cloud.stream.binder.LastSubscriberAwareChannel errorChannel,
                                                                                     boolean defaultErrorChannelPresent)

Return the default error message handler, which throws the error message payload to the caller if there are no user handlers subscribed. The handler is ordered so it runs after any user-defined handlers that are subscribed.

Parameters:

errorChannel - the error channel.

defaultErrorChannelPresent - true if the context has a default 'errorChannel'.

Returns:

the handler.

getErrorMessageStrategy

protected org.springframework.integration.support.ErrorMessageStrategy getErrorMessageStrategy()

Binders can return an ErrorMessageStrategy for building error messages; binder implementations typically might add extra headers to the error message.

Returns:

the implementation - may be null.

getErrorRecovererName

protected String getErrorRecovererName(ConsumerDestination destination,
                                       String group,
                                       C consumerProperties)

getErrorMessageHandlerName

protected String getErrorMessageHandlerName(ConsumerDestination destination,
                                            String group,
                                            C consumerProperties)

getErrorBridgeName

protected String getErrorBridgeName(ConsumerDestination destination,
                                    String group,
                                    C consumerProperties)

errorsBaseName

protected String errorsBaseName(ConsumerDestination destination,
                                String group,
                                C consumerProperties)

getErrorBridgeName

protected String getErrorBridgeName(ProducerDestination destination)

errorsBaseName

protected String errorsBaseName(ProducerDestination destination)