org.springframework.xd.dirt.integration.bus

Class MessageBusSupport

java.lang.Object

org.springframework.xd.dirt.integration.bus.MessageBusSupport

All Implemented Interfaces:

org.springframework.beans.factory.Aware, org.springframework.beans.factory.InitializingBean, org.springframework.context.ApplicationContextAware, MessageBus

Direct Known Subclasses:

KafkaMessageBus, LocalMessageBus, RabbitMessageBus, RedisMessageBus

public abstract class MessageBusSupport
extends java.lang.Object
implements MessageBus, org.springframework.context.ApplicationContextAware, org.springframework.beans.factory.InitializingBean

Author:

David Turanski, Gary Russell, Ilayaperumal Gopinathan

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	MessageBusSupport.DirectHandler
protected static class	MessageBusSupport.PartitioningMetadata
static class	MessageBusSupport.SetBuilder
protected class	MessageBusSupport.SharedChannelProvider<T extends org.springframework.messaging.MessageChannel> Looks up or optionally creates a new channel to use.

Nested classes/interfaces inherited from interface org.springframework.xd.dirt.integration.bus.MessageBus

MessageBus.Capability

Field Summary

Fields

Modifier and Type	Field and Description
protected static java.util.Set<java.lang.Object>	CONSUMER_RETRY_PROPERTIES
protected static java.util.Set<java.lang.Object>	CONSUMER_STANDARD_PROPERTIES The set of properties every bus implementation must support (or at least tolerate).
protected long	defaultBackOffInitialInterval
protected long	defaultBackOffMaxInterval
protected double	defaultBackOffMultiplier
protected int	defaultBatchBufferLimit
protected boolean	defaultBatchingEnabled
protected int	defaultBatchSize
protected long	defaultBatchTimeout
protected boolean	defaultCompress
protected int	defaultConcurrency
protected boolean	defaultDurableSubscription
protected int	defaultMaxAttempts
protected MessageBusSupport.SharedChannelProvider<org.springframework.integration.channel.DirectChannel>	directChannelProvider Used in the canonical case, when the binding does not involve an alias name.
protected org.springframework.expression.EvaluationContext	evaluationContext
protected static java.lang.String	JOB_CHANNEL_TYPE_PREFIX
protected org.slf4j.Logger	logger
protected static java.util.List<org.springframework.util.MimeType>	MEDIATYPES_MEDIATYPE_ALL
protected static java.lang.String	P2P_NAMED_CHANNEL_TYPE_PREFIX
protected static java.lang.String	PARTITION_HEADER
protected static java.util.Set<java.lang.Object>	PRODUCER_BATCHING_ADVANCED_PROPERTIES
protected static java.util.Set<java.lang.Object>	PRODUCER_BATCHING_BASIC_PROPERTIES
protected static java.util.Set<java.lang.Object>	PRODUCER_PARTITIONING_PROPERTIES
protected static java.util.Set<java.lang.Object>	PRODUCER_STANDARD_PROPERTIES
protected static java.lang.String	PUBSUB_NAMED_CHANNEL_TYPE_PREFIX
protected static java.lang.String	TAP_TYPE_PREFIX

Constructor Summary

Constructors

Constructor and Description
MessageBusSupport()

Method Summary

Methods

Modifier and Type	Method and Description
protected void	addBinding(Binding binding)
void	afterPropertiesSet()
static java.lang.String	applyPrefix(java.lang.String prefix, java.lang.String name) For bus implementations that support a prefix, apply the prefix to the name.
static java.lang.String	applyPubSub(java.lang.String name) For bus implementations that include a pub/sub component in identifiers, construct the name.
static java.lang.String	applyRequests(java.lang.String name) Build the requests entity name.
org.springframework.messaging.MessageChannel	bindDynamicProducer(java.lang.String name, java.util.Properties properties) Dynamically create a producer for the named channel.
org.springframework.messaging.MessageChannel	bindDynamicPubSubProducer(java.lang.String name, java.util.Properties properties) Dynamically create a producer for the named channel.
protected void	bindExistingProducerDirectlyIfPossible(java.lang.String name, org.springframework.messaging.MessageChannel consumerChannel) Attempt to bind a producer directly (avoiding the bus) if there is already a local producer.
protected boolean	bindNewProducerDirectlyIfPossible(java.lang.String name, org.springframework.messaging.SubscribableChannel moduleOutputChannel, AbstractBusPropertiesAccessor properties) Attempt to create a direct binding (avoiding the bus) if the consumer is local.
protected java.lang.String	buildPartitionRoutingExpression(java.lang.String expressionRoot)
protected org.springframework.retry.support.RetryTemplate	buildRetryTemplateIfRetryEnabled(AbstractBusPropertiesAccessor properties) Create and configure a retry template if the consumer 'maxAttempts' property is set.
static java.lang.String	constructDLQName(java.lang.String name) For bus implementations that support dead lettering, construct the name of the dead letter entity for the underlying pipe name.
protected void	deleteBinding(java.lang.String name, org.springframework.messaging.MessageChannel channel)
protected void	deleteBindings(java.lang.String name)
protected MessageValues	deserializePayloadIfNecessary(org.springframework.messaging.Message<?> message)
protected MessageValues	deserializePayloadIfNecessary(MessageValues message)
protected int	determinePartition(org.springframework.messaging.Message<?> message, MessageBusSupport.PartitioningMetadata meta) Determine the partition to which to send this message.
protected org.springframework.messaging.MessageChannel	doBindDynamicProducer(java.lang.String name, java.lang.String channelName, java.util.Properties properties) Create a producer for the named channel and bind it to the bus.
protected org.springframework.messaging.MessageChannel	doBindDynamicPubSubProducer(java.lang.String name, java.lang.String channelName, java.util.Properties properties) Create a producer for the named channel and bind it to the bus.
void	doManualAck(java.util.LinkedList<org.springframework.messaging.MessageHeaders> messageHeaders) Perform manual acknowledgement based on the metadata stored in message bus.
protected org.springframework.context.support.AbstractApplicationContext	getApplicationContext()
protected org.springframework.beans.factory.config.ConfigurableListableBeanFactory	getBeanFactory()
protected org.springframework.util.IdGenerator	getIdGenerator()
boolean	isCapable(MessageBus.Capability capability) Return true if the bus supports the capability.
protected boolean	isNamedChannel(java.lang.String name)
protected void	onInit()
protected MessageValues	serializePayloadIfNecessary(org.springframework.messaging.Message<?> message)
void	setApplicationContext(org.springframework.context.ApplicationContext applicationContext)
void	setCodec(org.springframework.integration.codec.Codec codec)
void	setDefaultBackOffInitialInterval(long defaultBackOffInitialInterval) Set the default retry back off initial interval for this bus; can be overridden with consumer 'backOffInitialInterval' property.
void	setDefaultBackOffMaxInterval(long defaultBackOffMaxInterval) Set the default retry back off max interval for this bus; can be overridden with consumer 'backOffMaxInterval' property.
void	setDefaultBackOffMultiplier(double defaultBackOffMultiplier) Set the default retry back off multiplier for this bus; can be overridden with consumer 'backOffMultiplier' property.
void	setDefaultBatchBufferLimit(int defaultBatchBufferLimit) Set the default batch buffer limit - used to send a batch early if its size exceeds this.
void	setDefaultBatchingEnabled(boolean defaultBatchingEnabled) Set whether this bus batches message sends by default.
void	setDefaultBatchSize(int defaultBatchSize) Set the default batch size; only applies when batching is enabled and the bus supports batching.
void	setDefaultBatchTimeout(long defaultBatchTimeout) Set the default batch timeout - used to send a batch if no messages arrive during this time.
void	setDefaultCompress(boolean defaultCompress) Set whether compression will be used by producers, by default.
void	setDefaultConcurrency(int defaultConcurrency) Set the default concurrency for this bus; can be overridden with consumer 'concurrency' property.
void	setDefaultDurableSubscription(boolean defaultDurableSubscription) Set whether subscriptions to taps/topics are durable.
void	setDefaultMaxAttempts(int defaultMaxAttempts) The default maximum delivery attempts for this bus.
void	setIntegrationEvaluationContext(org.springframework.expression.EvaluationContext evaluationContext)
void	setPartitionSelector(PartitionSelectorStrategy partitionSelector) Set the partition strategy to be used by this bus if no partitionExpression is provided for a module.
protected void	stopBindings()
void	unbindConsumer(java.lang.String name, org.springframework.messaging.MessageChannel channel) Unbind a specific p2p or pub/sub message consumer
void	unbindConsumers(java.lang.String name) Unbind an inbound inter-module channel and stop any active components that use the channel.
void	unbindProducer(java.lang.String name, org.springframework.messaging.MessageChannel channel) Unbind a specific p2p or pub/sub message producer
void	unbindProducers(java.lang.String name) Unbind an outbound inter-module channel and stop any active components that use the channel.
protected void	validateConsumerProperties(java.lang.String name, java.util.Properties properties, java.util.Set<java.lang.Object> supported) Validate the provided deployment properties for the consumer against those supported by this bus implementation.
protected void	validateProducerProperties(java.lang.String name, java.util.Properties properties, java.util.Set<java.lang.Object> supported) Validate the provided deployment properties for the producer against those supported by this bus implementation.

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.springframework.xd.dirt.integration.bus.MessageBus

bindConsumer, bindProducer, bindPubSubConsumer, bindPubSubProducer, bindReplier, bindRequestor

Field Detail

P2P_NAMED_CHANNEL_TYPE_PREFIX

protected static final java.lang.String P2P_NAMED_CHANNEL_TYPE_PREFIX

See Also:

Constant Field Values

TAP_TYPE_PREFIX

protected static final java.lang.String TAP_TYPE_PREFIX

See Also:

Constant Field Values

PUBSUB_NAMED_CHANNEL_TYPE_PREFIX

protected static final java.lang.String PUBSUB_NAMED_CHANNEL_TYPE_PREFIX

See Also:

Constant Field Values

JOB_CHANNEL_TYPE_PREFIX

protected static final java.lang.String JOB_CHANNEL_TYPE_PREFIX

See Also:

Constant Field Values

PARTITION_HEADER

protected static final java.lang.String PARTITION_HEADER

See Also:

Constant Field Values

logger

protected final org.slf4j.Logger logger

MEDIATYPES_MEDIATYPE_ALL

protected static final java.util.List<org.springframework.util.MimeType> MEDIATYPES_MEDIATYPE_ALL

CONSUMER_STANDARD_PROPERTIES

protected static final java.util.Set<java.lang.Object> CONSUMER_STANDARD_PROPERTIES

The set of properties every bus implementation must support (or at least tolerate).

PRODUCER_STANDARD_PROPERTIES

protected static final java.util.Set<java.lang.Object> PRODUCER_STANDARD_PROPERTIES

CONSUMER_RETRY_PROPERTIES

protected static final java.util.Set<java.lang.Object> CONSUMER_RETRY_PROPERTIES

PRODUCER_PARTITIONING_PROPERTIES

protected static final java.util.Set<java.lang.Object> PRODUCER_PARTITIONING_PROPERTIES

PRODUCER_BATCHING_BASIC_PROPERTIES

protected static final java.util.Set<java.lang.Object> PRODUCER_BATCHING_BASIC_PROPERTIES

PRODUCER_BATCHING_ADVANCED_PROPERTIES

protected static final java.util.Set<java.lang.Object> PRODUCER_BATCHING_ADVANCED_PROPERTIES

evaluationContext

protected volatile org.springframework.expression.EvaluationContext evaluationContext

directChannelProvider

protected final MessageBusSupport.SharedChannelProvider<org.springframework.integration.channel.DirectChannel> directChannelProvider

Used in the canonical case, when the binding does not involve an alias name.

defaultBackOffInitialInterval

protected volatile long defaultBackOffInitialInterval

defaultBackOffMaxInterval

protected volatile long defaultBackOffMaxInterval

defaultBackOffMultiplier

protected volatile double defaultBackOffMultiplier

defaultConcurrency

protected volatile int defaultConcurrency

defaultMaxAttempts

protected volatile int defaultMaxAttempts

defaultBatchingEnabled

protected volatile boolean defaultBatchingEnabled

defaultBatchSize

protected volatile int defaultBatchSize

defaultBatchBufferLimit

protected volatile int defaultBatchBufferLimit

defaultBatchTimeout

protected volatile long defaultBatchTimeout

defaultCompress

protected volatile boolean defaultCompress

defaultDurableSubscription

protected volatile boolean defaultDurableSubscription

Constructor Detail

MessageBusSupport

public MessageBusSupport()

Method Detail

applyPrefix

public static java.lang.String applyPrefix(java.lang.String prefix,
                           java.lang.String name)

For bus implementations that support a prefix, apply the prefix to the name.

Parameters:

prefix - the prefix.

name - the name.

applyPubSub

public static java.lang.String applyPubSub(java.lang.String name)

For bus implementations that include a pub/sub component in identifiers, construct the name.

Parameters:

name - the name.

applyRequests

public static java.lang.String applyRequests(java.lang.String name)

Build the requests entity name.

Parameters:

name - the name.

Returns:

the request entity name.

constructDLQName

public static java.lang.String constructDLQName(java.lang.String name)

For bus implementations that support dead lettering, construct the name of the dead letter entity for the underlying pipe name.

Parameters:

name - the name.

setApplicationContext

public void setApplicationContext(org.springframework.context.ApplicationContext applicationContext)
                           throws org.springframework.beans.BeansException

Specified by:

setApplicationContext in interface org.springframework.context.ApplicationContextAware

Throws:

org.springframework.beans.BeansException

getApplicationContext

protected org.springframework.context.support.AbstractApplicationContext getApplicationContext()

getBeanFactory

protected org.springframework.beans.factory.config.ConfigurableListableBeanFactory getBeanFactory()

setCodec

public void setCodec(org.springframework.integration.codec.Codec codec)

getIdGenerator

protected org.springframework.util.IdGenerator getIdGenerator()

setIntegrationEvaluationContext

public void setIntegrationEvaluationContext(org.springframework.expression.EvaluationContext evaluationContext)

setPartitionSelector

public void setPartitionSelector(PartitionSelectorStrategy partitionSelector)

Set the partition strategy to be used by this bus if no partitionExpression is provided for a module.

Parameters:

partitionSelector - The selector.

setDefaultBackOffInitialInterval

public void setDefaultBackOffInitialInterval(long defaultBackOffInitialInterval)

Set the default retry back off initial interval for this bus; can be overridden with consumer 'backOffInitialInterval' property.

Parameters:

defaultBackOffInitialInterval -

setDefaultBackOffMultiplier

public void setDefaultBackOffMultiplier(double defaultBackOffMultiplier)

Set the default retry back off multiplier for this bus; can be overridden with consumer 'backOffMultiplier' property.

Parameters:

defaultBackOffMultiplier -

setDefaultBackOffMaxInterval

public void setDefaultBackOffMaxInterval(long defaultBackOffMaxInterval)

Set the default retry back off max interval for this bus; can be overridden with consumer 'backOffMaxInterval' property.

Parameters:

defaultBackOffMaxInterval -

setDefaultConcurrency

public void setDefaultConcurrency(int defaultConcurrency)

Set the default concurrency for this bus; can be overridden with consumer 'concurrency' property.

Parameters:

defaultConcurrency -

setDefaultMaxAttempts

public void setDefaultMaxAttempts(int defaultMaxAttempts)

The default maximum delivery attempts for this bus. Can be overridden by consumer property 'maxAttempts' if supported. Values less than 2 disable retry and one delivery attempt is made.

Parameters:

defaultMaxAttempts - The default maximum attempts.

setDefaultBatchingEnabled

public void setDefaultBatchingEnabled(boolean defaultBatchingEnabled)

Set whether this bus batches message sends by default. Only applies to bus implementations that support batching.

Parameters:

defaultBatchingEnabled - the defaultBatchingEnabled to set.

setDefaultBatchSize

public void setDefaultBatchSize(int defaultBatchSize)

Set the default batch size; only applies when batching is enabled and the bus supports batching.

Parameters:

defaultBatchSize - the defaultBatchSize to set.

setDefaultBatchBufferLimit

public void setDefaultBatchBufferLimit(int defaultBatchBufferLimit)

Set the default batch buffer limit - used to send a batch early if its size exceeds this. Only applies if batching is enabled and the bus supports this property.

Parameters:

defaultBatchBufferLimit - the defaultBatchBufferLimit to set.

setDefaultBatchTimeout

public void setDefaultBatchTimeout(long defaultBatchTimeout)

Set the default batch timeout - used to send a batch if no messages arrive during this time. Only applies if batching is enabled and the bus supports this property.

Parameters:

defaultBatchTimeout - the defaultBatchTimeout to set.

setDefaultCompress

public void setDefaultCompress(boolean defaultCompress)

Set whether compression will be used by producers, by default.

Parameters:

defaultCompress - 'true' to use compression.

setDefaultDurableSubscription

public void setDefaultDurableSubscription(boolean defaultDurableSubscription)

Set whether subscriptions to taps/topics are durable.

Parameters:

defaultDurableSubscription - true for durable (default false).

afterPropertiesSet

public void afterPropertiesSet()
                        throws java.lang.Exception

Specified by:

afterPropertiesSet in interface org.springframework.beans.factory.InitializingBean

Throws:

java.lang.Exception

onInit

protected void onInit()

bindDynamicProducer

public org.springframework.messaging.MessageChannel bindDynamicProducer(java.lang.String name,
                                                               java.util.Properties properties)

Dynamically create a producer for the named channel.

Specified by:

bindDynamicProducer in interface MessageBus

Parameters:

name - The name.

properties - The properties.

Returns:

The channel.

doBindDynamicProducer

protected org.springframework.messaging.MessageChannel doBindDynamicProducer(java.lang.String name,
                                                                 java.lang.String channelName,
                                                                 java.util.Properties properties)

Create a producer for the named channel and bind it to the bus. Synchronized to avoid creating multiple instances.

Parameters:

name - The name.

channelName - The name of the channel to be created, and registered as bean.

properties - The properties.

Returns:

The channel.

bindDynamicPubSubProducer

public org.springframework.messaging.MessageChannel bindDynamicPubSubProducer(java.lang.String name,
                                                                     java.util.Properties properties)

Dynamically create a producer for the named channel. Note: even though it's pub/sub, we still use a direct channel. It will be bridged to a pub/sub channel in the local bus and bound to an appropriate element for other buses.

Specified by:

bindDynamicPubSubProducer in interface MessageBus

Parameters:

name - The name.

properties - The properties.

Returns:

The channel.

doBindDynamicPubSubProducer

protected org.springframework.messaging.MessageChannel doBindDynamicPubSubProducer(java.lang.String name,
                                                                       java.lang.String channelName,
                                                                       java.util.Properties properties)

Create a producer for the named channel and bind it to the bus. Synchronized to avoid creating multiple instances.

Parameters:

name - The name.

channelName - The name of the channel to be created, and registered as bean.

properties - The properties.

Returns:

The channel.

unbindConsumers

public void unbindConsumers(java.lang.String name)

Description copied from interface: MessageBus

Unbind an inbound inter-module channel and stop any active components that use the channel.

Specified by:

unbindConsumers in interface MessageBus

Parameters:

name - the channel name

unbindProducers

public void unbindProducers(java.lang.String name)

Description copied from interface: MessageBus

Unbind an outbound inter-module channel and stop any active components that use the channel.

Specified by:

unbindProducers in interface MessageBus

Parameters:

name - the channel name

unbindConsumer

public void unbindConsumer(java.lang.String name,
                  org.springframework.messaging.MessageChannel channel)

Description copied from interface: MessageBus

Unbind a specific p2p or pub/sub message consumer

Specified by:

unbindConsumer in interface MessageBus

Parameters:

name - The logical identify of a message source

channel - The channel bound as a consumer

unbindProducer

public void unbindProducer(java.lang.String name,
                  org.springframework.messaging.MessageChannel channel)

Description copied from interface: MessageBus

Unbind a specific p2p or pub/sub message producer

Specified by:

unbindProducer in interface MessageBus

Parameters:

name - the logical identity of the message target

channel - the channel bound as a producer

isCapable

public boolean isCapable(MessageBus.Capability capability)

Description copied from interface: MessageBus

Return true if the bus supports the capability.

Specified by:

isCapable in interface MessageBus

Parameters:

capability - the capability.

Returns:

true if the capability is supported.

addBinding

protected void addBinding(Binding binding)

deleteBindings

protected void deleteBindings(java.lang.String name)

deleteBinding

protected void deleteBinding(java.lang.String name,
                 org.springframework.messaging.MessageChannel channel)

stopBindings

protected void stopBindings()

serializePayloadIfNecessary

protected final MessageValues serializePayloadIfNecessary(org.springframework.messaging.Message<?> message)

deserializePayloadIfNecessary

protected final MessageValues deserializePayloadIfNecessary(org.springframework.messaging.Message<?> message)

deserializePayloadIfNecessary

protected final MessageValues deserializePayloadIfNecessary(MessageValues message)

determinePartition

protected int determinePartition(org.springframework.messaging.Message<?> message,
                     MessageBusSupport.PartitioningMetadata meta)

Determine the partition to which to send this message. If a partition key extractor class is provided, it is invoked to determine the key. Otherwise, the partition key expression is evaluated to obtain the key value. If a partition selector class is provided, it will be invoked to determine the partition. Otherwise, if the partition expression is not null, it is evaluated against the key and is expected to return an integer to which the modulo function will be applied, using the partitionCount as the divisor. If no partition expression is provided, the key will be passed to the bus partition strategy along with the partitionCount. The default partition strategy uses key.hashCode(), and the result will be the mod of that value.

Parameters:

message - the message.

meta - the partitioning metadata.

Returns:

the partition.

validateConsumerProperties

protected void validateConsumerProperties(java.lang.String name,
                              java.util.Properties properties,
                              java.util.Set<java.lang.Object> supported)

Validate the provided deployment properties for the consumer against those supported by this bus implementation. The consumer is that part of the bus that consumes messages from the underlying infrastructure and sends them to the next module. Consumer properties are used to configure the consumer.

Parameters:

name - The name.

properties - The properties.

supported - The supported properties.

validateProducerProperties

protected void validateProducerProperties(java.lang.String name,
                              java.util.Properties properties,
                              java.util.Set<java.lang.Object> supported)

Validate the provided deployment properties for the producer against those supported by this bus implementation. When a module sends a message to the bus, the producer uses these properties while sending it to the underlying infrastructure.

Parameters:

name - The name.

properties - The properties.

supported - The supported properties.

buildPartitionRoutingExpression

protected java.lang.String buildPartitionRoutingExpression(java.lang.String expressionRoot)

buildRetryTemplateIfRetryEnabled

protected org.springframework.retry.support.RetryTemplate buildRetryTemplateIfRetryEnabled(AbstractBusPropertiesAccessor properties)

Create and configure a retry template if the consumer 'maxAttempts' property is set.

Parameters:

properties - The properties.

Returns:

The retry template, or null if retry is not enabled.

isNamedChannel

protected boolean isNamedChannel(java.lang.String name)

bindNewProducerDirectlyIfPossible

protected boolean bindNewProducerDirectlyIfPossible(java.lang.String name,
                                        org.springframework.messaging.SubscribableChannel moduleOutputChannel,
                                        AbstractBusPropertiesAccessor properties)

Attempt to create a direct binding (avoiding the bus) if the consumer is local. Named channel producers are not bound directly.

Parameters:

name - The name.

moduleOutputChannel - The channel to bind.

properties - The producer properties.

Returns:

true if the producer is bound.

bindExistingProducerDirectlyIfPossible

protected void bindExistingProducerDirectlyIfPossible(java.lang.String name,
                                          org.springframework.messaging.MessageChannel consumerChannel)

Attempt to bind a producer directly (avoiding the bus) if there is already a local producer. PubSub producers cannot be bound directly. Create the direct binding, then unbind the existing bus producer.

Parameters:

name - The name.

consumerChannel - The channel to bind the producer to.

doManualAck

public void doManualAck(java.util.LinkedList<org.springframework.messaging.MessageHeaders> messageHeaders)

Perform manual acknowledgement based on the metadata stored in message bus.