org.springframework.xd.reactor

Class BroadcasterMessageHandler

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

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 BroadcasterMessageHandler
extends AbstractReactorMessageHandler

Adapts the item at a time delivery of a MessageHandler by delegating processing to a Stream

The outputStream of the processor is used to create a message and send it to the output channel. If the input channel and output channel are connected to the MessageBus, then data delivered to the input stream via a call to onNext is invoked on the dispatcher thread of the message bus and sending a message to the output channel will involve IO operations on the message bus.

The implementation uses a RingBufferProcessor with asynchronous dispatch. This has the advantage that the state of the Stream can be shared across all the incoming dispatcher threads that are invoking onNext. It has the disadvantage that processing and sending to the output channel will execute serially on one of the dispatcher threads.

The use of this handler makes for a very natural first experience when processing data. For example given the stream http | reactor-processor | log where the reactor-processor does does a buffer(5) and then produces a single value. Sending 10 messages to the http source will result in 2 messages in the log, no matter how many dispatcher threads are used.

You can modify what thread the outputStream subscriber, which does the send to the output channel, will use by explicitly calling dispatchOn or other switch (http://projectreactor.io/docs/reference/#streams-multithreading) before returning the outputStream from your processor.

Use MultipleBroadcasterMessageHandler for concurrent execution on dispatcher threads spread across across multiple Stream.

All error handling is the responsibility of the processor implementation.

Author:

Mark Pollack, Stephane Malbdini

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
BroadcasterMessageHandler(Processor processor) Construct a new BroadcasterMessageHandler given the reactor based Processor to delegate processing to.

Method Summary

Methods

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

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

BroadcasterMessageHandler

public BroadcasterMessageHandler(Processor processor)

Construct a new BroadcasterMessageHandler given the reactor based Processor to delegate processing to.

Parameters:

processor - The stream based reactor processor

Method Detail

handleMessageInternal

protected void handleMessageInternal(org.springframework.messaging.Message<?> message)
                              throws java.lang.Exception

Specified by:

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

Throws:

java.lang.Exception

destroy

public void destroy()
             throws java.lang.Exception

Throws:

java.lang.Exception

onInit

protected void onInit()
               throws java.lang.Exception

Overrides:

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

Throws:

java.lang.Exception