org.springframework.xd.reactor

Class MultipleBroadcasterMessageHandler

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.handler.AbstractMessageHandler

org.springframework.integration.handler.AbstractMessageProducingHandler

org.springframework.xd.reactor.AbstractReactorMessageHandler

org.springframework.xd.reactor.MultipleBroadcasterMessageHandler

All Implemented Interfaces:

org.springframework.beans.factory.Aware, org.springframework.beans.factory.BeanFactoryAware, org.springframework.beans.factory.BeanNameAware, org.springframework.beans.factory.DisposableBean, org.springframework.beans.factory.InitializingBean, org.springframework.context.ApplicationContextAware, org.springframework.core.Ordered, org.springframework.integration.context.Orderable, org.springframework.integration.core.MessageProducer, org.springframework.integration.support.context.NamedComponent, org.springframework.integration.support.management.ConfigurableMetricsAware<org.springframework.integration.support.management.AbstractMessageHandlerMetrics>, org.springframework.integration.support.management.IntegrationManagement, org.springframework.integration.support.management.IntegrationStatsManagement, org.springframework.integration.support.management.MessageHandlerMetrics, org.springframework.integration.support.management.TrackableComponent, org.springframework.messaging.MessageHandler

public class MultipleBroadcasterMessageHandler
extends AbstractReactorMessageHandler

Adapts the item at a time delivery of a MessageHandler by delegating processing to a Stream based on a partitionExpression.

The specific Stream that the message is delegated to is determined by the partitionExpression value. Unless you change the scheduling of the inputStream in your processor, you should ensure that the partitionExpression does not map messages delivered on different message bus dispatcher threads to the same stream. This is due to the underlying use of a Broadcaster.

For example, using the expression T(java.lang.Thread).currentThread().getId() would map the current dispatcher thread id to an instance of a Stream. If you wanted to have a Stream per Kafka partition, you can use the expression header['kafka_partition_id'] since the MessageBus dispatcher thread will be the same for each partition.

If the Stream mapped to the partitionExpression value has an error or completes, it will be recreated when the next message consumed maps to the same partitionExpression value.

All error handling is the responsibility of the processor implementation.

Author:

Mark Pollack, Stephane Maldini, Gary Russell

Nested Class Summary

Nested classes/interfaces inherited from class org.springframework.xd.reactor.AbstractReactorMessageHandler

AbstractReactorMessageHandler.ChannelForwardingSubscriber

Field Summary

Fields inherited from class org.springframework.xd.reactor.AbstractReactorMessageHandler

logger, processor

Fields inherited from class org.springframework.integration.handler.AbstractMessageProducingHandler

messagingTemplate

Fields inherited from interface org.springframework.core.Ordered

HIGHEST_PRECEDENCE, LOWEST_PRECEDENCE

Constructor Summary

Constructors

Constructor and Description
MultipleBroadcasterMessageHandler(Processor processor, java.lang.String partitionExpression) Construct a new MessageHandler given the reactor based Processor to delegate processing to and a partition expression.

Method Summary

Methods

Modifier and Type	Method and Description
void	destroy()
protected void	handleMessageInternal(org.springframework.messaging.Message<?> message)
protected void	onInit()
void	setIntegrationEvaluationContext(org.springframework.expression.EvaluationContext evaluationContext)

Methods inherited from class org.springframework.xd.reactor.AbstractReactorMessageHandler

getEnvironment, getInputType, getRingBufferSize, getStopTimeout, invokeProcessor, setRingBufferSize, setStopTimeout

Methods inherited from class org.springframework.integration.handler.AbstractMessageProducingHandler

getOutputChannel, produceOutput, sendOutputs, setOutputChannel, setOutputChannelName, setSendTimeout, shouldCopyRequestHeaders, shouldSplitOutput

Methods inherited from class org.springframework.integration.handler.AbstractMessageHandler

configureMetrics, getActiveCount, getActiveCountLong, getComponentType, getDuration, getErrorCount, getErrorCountLong, getHandleCount, getHandleCountLong, getManagedName, getManagedType, getMaxDuration, getMeanDuration, getMinDuration, getOrder, getStandardDeviationDuration, handleMessage, isCountsEnabled, isLoggingEnabled, isStatsEnabled, reset, setCountsEnabled, setLoggingEnabled, setManagedName, setManagedType, setOrder, setShouldTrack, setStatsEnabled

Methods inherited from class org.springframework.integration.context.IntegrationObjectSupport

afterPropertiesSet, extractTypeIfPossible, getApplicationContext, getApplicationContextId, getBeanFactory, getChannelResolver, getComponentName, getConversionService, getIntegrationProperties, getIntegrationProperty, getMessageBuilderFactory, getTaskScheduler, setApplicationContext, setBeanFactory, setBeanName, setChannelResolver, setComponentName, setConversionService, setMessageBuilderFactory, setTaskScheduler, toString

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.springframework.integration.support.context.NamedComponent

getComponentName

Constructor Detail

MultipleBroadcasterMessageHandler

public MultipleBroadcasterMessageHandler(Processor processor,
                                 java.lang.String partitionExpression)

Construct a new MessageHandler given the reactor based Processor to delegate processing to and a partition expression.

Parameters:

processor - The stream based reactor processor

Method Detail

setIntegrationEvaluationContext

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

onInit

protected void onInit()
               throws java.lang.Exception

Overrides:

onInit in class org.springframework.integration.handler.AbstractMessageProducingHandler

Throws:

java.lang.Exception

handleMessageInternal

protected void handleMessageInternal(org.springframework.messaging.Message<?> message)

Specified by:

handleMessageInternal in class org.springframework.integration.handler.AbstractMessageHandler

destroy

public void destroy()
             throws java.lang.Exception

Throws:

java.lang.Exception