org.springframework.xd.dirt.integration.bus.local

Class LocalMessageBus

java.lang.Object

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

org.springframework.xd.dirt.integration.bus.local.LocalMessageBus

All Implemented Interfaces:

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

public class LocalMessageBus
extends MessageBusSupport

A simple implementation of MessageBus for in-process use. For inbound and outbound, creates a DirectChannel or a QueueChannel depending on whether the binding is aliased or not then bridges the passed MessageChannel to the channel which is registered in the given application context. If that channel does not yet exist, it will be created.

Since:

1.0

Author:

David Turanski, Mark Fisher, Gary Russell, Jennifer Hickey, Ilayaperumal Gopinathan

Nested Class Summary

Nested classes/interfaces inherited from class org.springframework.xd.dirt.integration.bus.MessageBusSupport

MessageBusSupport.DirectHandler, MessageBusSupport.PartitioningMetadata, MessageBusSupport.SetBuilder, MessageBusSupport.SharedChannelProvider<T extends org.springframework.messaging.MessageChannel>

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_REQUEST_REPLY_PROPERTIES

Fields inherited from class org.springframework.xd.dirt.integration.bus.MessageBusSupport

CONSUMER_RETRY_PROPERTIES, CONSUMER_STANDARD_PROPERTIES, defaultBackOffInitialInterval, defaultBackOffMaxInterval, defaultBackOffMultiplier, defaultBatchBufferLimit, defaultBatchingEnabled, defaultBatchSize, defaultBatchTimeout, defaultCompress, defaultConcurrency, defaultDurableSubscription, defaultMaxAttempts, directChannelProvider, evaluationContext, JOB_CHANNEL_TYPE_PREFIX, logger, MEDIATYPES_MEDIATYPE_ALL, P2P_NAMED_CHANNEL_TYPE_PREFIX, PARTITION_HEADER, PRODUCER_BATCHING_ADVANCED_PROPERTIES, PRODUCER_BATCHING_BASIC_PROPERTIES, PRODUCER_PARTITIONING_PROPERTIES, PRODUCER_STANDARD_PROPERTIES, PUBSUB_NAMED_CHANNEL_TYPE_PREFIX, TAP_TYPE_PREFIX

Constructor Summary

Constructors

Constructor and Description
LocalMessageBus()

Method Summary

Methods

Modifier and Type	Method and Description
void	bindConsumer(java.lang.String name, org.springframework.messaging.MessageChannel moduleInputChannel, java.util.Properties properties) Looks up or creates a DirectChannel with the given name and creates a bridge from that channel to the provided channel instance.
org.springframework.messaging.MessageChannel	bindDynamicProducer(java.lang.String name, java.util.Properties properties) For the local bus we bridge the router "output" channel to a queue channel; the queue channel gets the name and the source channel is named 'dynamic.output.to.' + name.
org.springframework.messaging.MessageChannel	bindDynamicPubSubProducer(java.lang.String name, java.util.Properties properties) For the local bus we bridge the router "output" channel to a pub/sub channel; the pub/sub channel gets the name and the source channel is named 'dynamic.output.to.' + name.
void	bindProducer(java.lang.String name, org.springframework.messaging.MessageChannel moduleOutputChannel, java.util.Properties properties) Looks up or creates a DirectChannel with the given name and creates a bridge to that channel from the provided channel instance.
void	bindPubSubConsumer(java.lang.String name, org.springframework.messaging.MessageChannel moduleInputChannel, java.util.Properties properties) Bind a message consumer on a pub/sub channel
void	bindPubSubProducer(java.lang.String name, org.springframework.messaging.MessageChannel moduleOutputChannel, java.util.Properties properties) Bind a message producer on a pub/sub channel.
void	bindReplier(java.lang.String name, org.springframework.messaging.MessageChannel requests, org.springframework.messaging.MessageChannel replies, java.util.Properties properties) Bind a consumer that handles requests from a requestor and asynchronously sends replies.
void	bindRequestor(java.lang.String name, org.springframework.messaging.MessageChannel requests, org.springframework.messaging.MessageChannel replies, java.util.Properties properties) Bind a producer that expects async replies.
protected org.springframework.integration.handler.BridgeHandler	bridge(java.lang.String name, org.springframework.messaging.MessageChannel from, org.springframework.messaging.MessageChannel to, java.lang.String bridgeName, java.util.Collection<org.springframework.util.MimeType> acceptedMimeTypes, org.springframework.xd.dirt.integration.bus.local.LocalMessageBus.LocalBusPropertiesAccessor properties)
protected org.springframework.integration.handler.BridgeHandler	bridge(java.lang.String name, org.springframework.messaging.MessageChannel from, org.springframework.messaging.MessageChannel to, java.lang.String bridgeName, org.springframework.xd.dirt.integration.bus.local.LocalMessageBus.LocalBusPropertiesAccessor properties)
protected <T> T	getBean(java.lang.String name, java.lang.Class<T> requiredType)
protected void	onInit()
void	setExecutorCorePoolSize(int executorCorePoolSize) Set the ThreadPoolTaskExecutor} core pool size to limit the number of concurrent threads.
void	setExecutorKeepAliveSeconds(int executorKeepAliveSeconds) Set the ThreadPoolTaskExecutor} keep alive seconds.
void	setExecutorMaxPoolSize(int executorMaxPoolSize) Set the ThreadPoolTaskExecutor} max pool size to limit the number of concurrent threads.
void	setExecutorQueueSize(int executorQueueSize) Set the ThreadPoolTaskExecutor} queue size to limit the number of concurrent threads.
void	setPoller(org.springframework.integration.scheduling.PollerMetadata poller) Set the poller to use when QueueChannels are used.
void	setQueueSize(int queueSize) Set the size of the queue when using QueueChannels.
void	unbindProducer(java.lang.String name, org.springframework.messaging.MessageChannel channel) Unbind a specific p2p or pub/sub message producer

Methods inherited from class org.springframework.xd.dirt.integration.bus.MessageBusSupport

addBinding, afterPropertiesSet, applyPrefix, applyPubSub, applyRequests, bindExistingProducerDirectlyIfPossible, bindNewProducerDirectlyIfPossible, buildPartitionRoutingExpression, buildRetryTemplateIfRetryEnabled, constructDLQName, deleteBinding, deleteBindings, deserializePayloadIfNecessary, deserializePayloadIfNecessary, determinePartition, doBindDynamicProducer, doBindDynamicPubSubProducer, doManualAck, getApplicationContext, getBeanFactory, getIdGenerator, isCapable, isNamedChannel, serializePayloadIfNecessary, setApplicationContext, setCodec, setDefaultBackOffInitialInterval, setDefaultBackOffMaxInterval, setDefaultBackOffMultiplier, setDefaultBatchBufferLimit, setDefaultBatchingEnabled, setDefaultBatchSize, setDefaultBatchTimeout, setDefaultCompress, setDefaultConcurrency, setDefaultDurableSubscription, setDefaultMaxAttempts, setIntegrationEvaluationContext, setPartitionSelector, stopBindings, unbindConsumer, unbindConsumers, unbindProducers, validateConsumerProperties, validateProducerProperties

Methods inherited from class java.lang.Object

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

Field Detail

CONSUMER_REQUEST_REPLY_PROPERTIES

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

Constructor Detail

LocalMessageBus

public LocalMessageBus()

Method Detail

setPoller

public void setPoller(org.springframework.integration.scheduling.PollerMetadata poller)

Set the poller to use when QueueChannels are used.

setQueueSize

public void setQueueSize(int queueSize)

Set the size of the queue when using QueueChannels.

setExecutorCorePoolSize

public void setExecutorCorePoolSize(int executorCorePoolSize)

Set the ThreadPoolTaskExecutor} core pool size to limit the number of concurrent threads. The executor is used for PubSub operations. Default: 0 (threads created on demand until maxPoolSize).

Parameters:

executorCorePoolSize - the pool size.

setExecutorMaxPoolSize

public void setExecutorMaxPoolSize(int executorMaxPoolSize)

Set the ThreadPoolTaskExecutor} max pool size to limit the number of concurrent threads. The executor is used for PubSub operations. Default: 200.

Parameters:

executorMaxPoolSize - the pool size.

setExecutorQueueSize

public void setExecutorQueueSize(int executorQueueSize)

Set the ThreadPoolTaskExecutor} queue size to limit the number of concurrent threads. The executor is used for PubSub operations. Default: Integer.MAX_VALUE.

Parameters:

executorQueueSize - the queue size.

setExecutorKeepAliveSeconds

public void setExecutorKeepAliveSeconds(int executorKeepAliveSeconds)

Set the ThreadPoolTaskExecutor} keep alive seconds. The executor is used for PubSub operations.

Parameters:

executorKeepAliveSeconds - the keep alive seconds.

onInit

protected void onInit()

Overrides:

onInit in class MessageBusSupport

bindDynamicProducer

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

For the local bus we bridge the router "output" channel to a queue channel; the queue channel gets the name and the source channel is named 'dynamic.output.to.' + name. Dynamically create a producer for the named channel.

Specified by:

bindDynamicProducer in interface MessageBus

Overrides:

bindDynamicProducer in class MessageBusSupport

Parameters:

name - The name.

properties - The properties.

Returns:

The channel.

bindDynamicPubSubProducer

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

For the local bus we bridge the router "output" channel to a pub/sub channel; the pub/sub channel gets the name and the source channel is named 'dynamic.output.to.' + name. 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

Overrides:

bindDynamicPubSubProducer in class MessageBusSupport

Parameters:

name - The name.

properties - The properties.

Returns:

The channel.

bindConsumer

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

Looks up or creates a DirectChannel with the given name and creates a bridge from that channel to the provided channel instance.

Parameters:

name - the logical identity of the message source

moduleInputChannel - the channel bound as a consumer

properties - arbitrary String key/value pairs that will be used in the binding

bindPubSubConsumer

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

Description copied from interface: MessageBus

Bind a message consumer on a pub/sub channel

Parameters:

name - the logical identity of the message source

moduleInputChannel - the channel bound as a pub/sub consumer

properties - arbitrary String key/value pairs that will be used in the binding

bindProducer

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

Looks up or creates a DirectChannel with the given name and creates a bridge to that channel from the provided channel instance.

Parameters:

name - the logical identity of the message target

moduleOutputChannel - the channel bound as a producer

properties - arbitrary String key/value pairs that will be used in the binding

bindPubSubProducer

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

Description copied from interface: MessageBus

Bind a message producer on a pub/sub channel.

Parameters:

name - the logical identity of the message target

moduleOutputChannel - the channel bound as a producer

properties - arbitrary String key/value pairs that will be used in the binding

bindRequestor

public void bindRequestor(java.lang.String name,
                 org.springframework.messaging.MessageChannel requests,
                 org.springframework.messaging.MessageChannel replies,
                 java.util.Properties properties)

Description copied from interface: MessageBus

Bind a producer that expects async replies. To unbind, invoke unbindProducer() and unbindConsumer().

Parameters:

name - The name of the requestor.

requests - The request channel - sends requests.

replies - The reply channel - receives replies.

properties - arbitrary String key/value pairs that will be used in the binding.

bindReplier

public void bindReplier(java.lang.String name,
               org.springframework.messaging.MessageChannel requests,
               org.springframework.messaging.MessageChannel replies,
               java.util.Properties properties)

Description copied from interface: MessageBus

Bind a consumer that handles requests from a requestor and asynchronously sends replies. To unbind, invoke unbindProducer() and unbindConsumer().

Parameters:

name - The name of the requestor for which this replier will handle requests.

requests - The request channel - receives requests.

replies - The reply channel - sends replies.

properties - arbitrary String key/value pairs that will be used in the binding.

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

Overrides:

unbindProducer in class MessageBusSupport

Parameters:

name - the logical identity of the message target

channel - the channel bound as a producer

bridge

protected org.springframework.integration.handler.BridgeHandler bridge(java.lang.String name,
                                                           org.springframework.messaging.MessageChannel from,
                                                           org.springframework.messaging.MessageChannel to,
                                                           java.lang.String bridgeName,
                                                           org.springframework.xd.dirt.integration.bus.local.LocalMessageBus.LocalBusPropertiesAccessor properties)

bridge

protected org.springframework.integration.handler.BridgeHandler bridge(java.lang.String name,
                                                           org.springframework.messaging.MessageChannel from,
                                                           org.springframework.messaging.MessageChannel to,
                                                           java.lang.String bridgeName,
                                                           java.util.Collection<org.springframework.util.MimeType> acceptedMimeTypes,
                                                           org.springframework.xd.dirt.integration.bus.local.LocalMessageBus.LocalBusPropertiesAccessor properties)

getBean

protected <T> T getBean(java.lang.String name,
            java.lang.Class<T> requiredType)