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, org.springframework.integration.expression.IntegrationEvaluationContextAware, MessageBus

Direct Known Subclasses:

LocalMessageBus, RabbitMessageBus, RedisMessageBus

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

Author:

David Turanski, Gary Russell

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	MessageBusSupport.DirectHandler
protected 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.

Field Summary

Fields

Modifier and Type	Field and Description
protected static java.util.Set<java.lang.Object>	CONSUMER_RETRY_PROPERTIES
protected long	defaultBackOffInitialInterval
protected long	defaultBackOffMaxInterval
protected double	defaultBackOffMultiplier
protected int	defaultConcurrency
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.apache.commons.logging.Log	logger
protected static java.util.List<org.springframework.util.MimeType>	MEDIATYPES_MEDIATYPE_ALL
protected static java.lang.String	ORIGINAL_CONTENT_TYPE_HEADER
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_PARTITIONING_PROPERTIES
protected static java.lang.String	PUBSUB_NAMED_CHANNEL_TYPE_PREFIX
protected MessageBusSupport.SharedChannelProvider<org.springframework.integration.channel.PublishSubscribeChannel>	pubsubChannelProvider
protected MessageBusSupport.SharedChannelProvider<org.springframework.integration.channel.QueueChannel>	queueChannelProvider Used to create and customize QueueChannels when the binding operation involves aliased names.

Constructor Summary

Constructors

Constructor and Description
MessageBusSupport()

Method Summary

Methods

Modifier and Type	Method and Description
protected void	addBinding(Binding binding)
void	afterPropertiesSet()
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.
protected void	deleteBinding(java.lang.String name, org.springframework.messaging.MessageChannel channel)
protected void	deleteBindings(java.lang.String name)
protected org.springframework.messaging.Message<?>	deserializePayloadIfNecessary(org.springframework.messaging.Message<?> 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.
protected org.springframework.context.support.AbstractApplicationContext	getApplicationContext()
protected org.springframework.beans.factory.config.ConfigurableListableBeanFactory	getBeanFactory()
protected org.springframework.util.IdGenerator	getIdGenerator()
protected boolean	isNamedChannel(java.lang.String name)
protected void	onInit()
protected org.springframework.messaging.Message<?>	serializePayloadIfNecessary(org.springframework.messaging.Message<?> message, org.springframework.util.MimeType to)
void	setApplicationContext(org.springframework.context.ApplicationContext applicationContext)
void	setCodec(MultiTypeCodec<java.lang.Object> 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	setDefaultConcurrency(int defaultConcurrency) Set the default concurrency for this bus; can be overridden with consumer 'concurrency' property.
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.
void	setQueueSize(int queueSize) Set the size of the queue when using QueueChannels.
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

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.apache.commons.logging.Log logger

ORIGINAL_CONTENT_TYPE_HEADER

protected static final java.lang.String ORIGINAL_CONTENT_TYPE_HEADER

See Also:

Constant Field Values

MEDIATYPES_MEDIATYPE_ALL

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

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

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.

queueChannelProvider

protected final MessageBusSupport.SharedChannelProvider<org.springframework.integration.channel.QueueChannel> queueChannelProvider

Used to create and customize QueueChannels when the binding operation involves aliased names.

pubsubChannelProvider

protected final MessageBusSupport.SharedChannelProvider<org.springframework.integration.channel.PublishSubscribeChannel> pubsubChannelProvider

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

Constructor Detail

MessageBusSupport

public MessageBusSupport()

Method Detail

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

setQueueSize

public void setQueueSize(int queueSize)

Set the size of the queue when using QueueChannels.

setCodec

public void setCodec(MultiTypeCodec<java.lang.Object> codec)

getIdGenerator

protected org.springframework.util.IdGenerator getIdGenerator()

setIntegrationEvaluationContext

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

Specified by:

setIntegrationEvaluationContext in interface org.springframework.integration.expression.IntegrationEvaluationContextAware

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.

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

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 org.springframework.messaging.Message<?> serializePayloadIfNecessary(org.springframework.messaging.Message<?> message,
                                                                   org.springframework.util.MimeType to)

deserializePayloadIfNecessary

protected final org.springframework.messaging.Message<?> deserializePayloadIfNecessary(org.springframework.messaging.Message<?> 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.