org.springframework.xd.rxjava

Class SubjectMessageHandler

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.handler.AbstractMessageHandler

org.springframework.integration.handler.AbstractMessageProducingHandler

org.springframework.xd.rxjava.SubjectMessageHandler

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.history.TrackableComponent, org.springframework.integration.support.context.NamedComponent, org.springframework.messaging.MessageHandler

public class SubjectMessageHandler
extends org.springframework.integration.handler.AbstractMessageProducingHandler
implements org.springframework.beans.factory.DisposableBean

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

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 SerializedSubject. This has the advantage that the state of the Observabale 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 | rxjava-processor | log where the rxjava-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 observeOn before returning the outputStream from your processor.

Use MultipleSubjectMessageHandler for concurrent execution on dispatcher threads spread across across multiple Observables. All error handling is the responsibility of the processor implementation.

Author:

Mark Pollack

Field Summary

Fields

Modifier and Type	Field and Description
protected org.slf4j.Logger	logger

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
SubjectMessageHandler(Processor processor)

Method Summary

Methods

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

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

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

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

getComponentType, getOrder, handleMessage, setOrder, setShouldTrack

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

Field Detail

logger

protected final org.slf4j.Logger logger

Constructor Detail

SubjectMessageHandler

public SubjectMessageHandler(Processor 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

Specified by:

destroy in interface org.springframework.beans.factory.DisposableBean

Throws:

java.lang.Exception