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>, org.springframework.context.ApplicationContextAware

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>

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

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

Since:

1.1

Author:

Marius Bogoevici, Ilayaperumal Gopinathan, Soby Chacko

Nested Class Summary

Nested Classes

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

Field Summary

Fields

Modifier and Type	Field and Description
protected static org.springframework.expression.ExpressionParser	EXPRESSION_PARSER
protected PP	provisioningProvider ProvisioningProvider delegated by the downstream binder implementations.

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

contentTypeResolver, logger

Constructor Summary

Constructors

Constructor and Description
AbstractMessageChannelBinder(boolean supportsHeadersNatively, String[] headersToEmbed, PP provisioningProvider)

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.
protected abstract org.springframework.integration.core.MessageProducer	createConsumerEndpoint(ConsumerDestination destination, String group, C properties) Creates MessageProducer that receives data from the consumer destination.
protected abstract org.springframework.messaging.MessageHandler	createProducerMessageHandler(ProducerDestination destination, P producerProperties, org.springframework.messaging.MessageChannel errorChannel) Creates 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.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 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.

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

afterPropertiesSet, applyPrefix, bindConsumer, bindProducer, buildPartitionRoutingExpression, buildRetryTemplate, constructDLQName, deserializePayloadIfNecessary, deserializePayloadIfNecessary, getApplicationContext, getBeanFactory, groupedName, onInit, serializePayloadIfNecessary, serializePayloadIfNecessary, setApplicationContext, setCodec, setIntegrationEvaluationContext

Methods inherited from class java.lang.Object

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

Field Detail

EXPRESSION_PARSER

protected static final org.springframework.expression.ExpressionParser EXPRESSION_PARSER

provisioningProvider

protected final PP extends ProvisioningProvider<C,P> provisioningProvider

ProvisioningProvider delegated by the downstream binder implementations.

Constructor Detail

AbstractMessageChannelBinder

public AbstractMessageChannelBinder(boolean supportsHeadersNatively,
                                    String[] headersToEmbed,
                                    PP provisioningProvider)

Method Detail

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) 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 e ndpoint 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

createProducerMessageHandler

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

Creates 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

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.

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

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)