org.springframework.cloud.stream.annotation.rxjava

Class SubjectMessageHandler

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.handler.AbstractMessageHandler

org.springframework.integration.handler.AbstractMessageProducingHandler

org.springframework.cloud.stream.annotation.rxjava.SubjectMessageHandler

All Implemented Interfaces:

org.springframework.beans.factory.Aware, org.springframework.beans.factory.BeanFactoryAware, org.springframework.beans.factory.BeanNameAware, org.springframework.beans.factory.InitializingBean, org.springframework.context.ApplicationContextAware, org.springframework.context.Lifecycle, org.springframework.context.Phased, org.springframework.context.SmartLifecycle, org.springframework.core.Ordered, org.springframework.integration.context.ExpressionCapable, org.springframework.integration.context.Orderable, org.springframework.integration.core.MessageProducer, org.springframework.integration.handler.HeaderPropagationAware, 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

Deprecated.

@Deprecated
public class SubjectMessageHandler
extends org.springframework.integration.handler.AbstractMessageProducingHandler
implements org.springframework.context.SmartLifecycle

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 Binder, then data delivered to the input stream via a call to onNext is invoked on the dispatcher thread of the binder and sending a message to the output channel will involve IO operations on the binder.

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

All error handling is the responsibility of the processor implementation.

Author:

Mark Pollack, Ilayaperumal Gopinathan, Marius Bogoevici

Field Summary

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

messagingTemplate

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

EXPRESSION_PARSER

Fields inherited from interface org.springframework.core.Ordered

HIGHEST_PRECEDENCE, LOWEST_PRECEDENCE

Constructor Summary

Constructors

Constructor and Description
SubjectMessageHandler(RxJavaProcessor processor) Deprecated.

Method Summary

Modifier and Type	Method and Description
int	getPhase() Deprecated.
protected void	handleMessageInternal(org.springframework.messaging.Message<?> message) Deprecated.
boolean	isAutoStartup() Deprecated.
boolean	isRunning() Deprecated.
void	start() Deprecated.
void	stop() Deprecated.
void	stop(Runnable callback) Deprecated.

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

addNotPropagatedHeaders, createOutputMessage, getNotPropagatedHeaders, getOutputChannel, isAsync, onInit, produceOutput, resolveErrorChannel, sendErrorMessage, sendOutput, sendOutputs, setAsync, setNotPropagatedHeaders, 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, getExpression, getIntegrationProperties, getIntegrationProperty, getMessageBuilderFactory, getTaskScheduler, setApplicationContext, setBeanFactory, setBeanName, setChannelResolver, setComponentName, setConversionService, setMessageBuilderFactory, setPrimaryExpression, 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

SubjectMessageHandler

public SubjectMessageHandler(RxJavaProcessor processor)

Deprecated.

Method Detail

start

public void start()

Deprecated.

Specified by:

start in interface org.springframework.context.Lifecycle

isRunning

public boolean isRunning()

Deprecated.

Specified by:

isRunning in interface org.springframework.context.Lifecycle

isAutoStartup

public boolean isAutoStartup()

Deprecated.

Specified by:

isAutoStartup in interface org.springframework.context.SmartLifecycle

stop

public void stop(Runnable callback)

Deprecated.

Specified by:

stop in interface org.springframework.context.SmartLifecycle

getPhase

public int getPhase()

Deprecated.

Specified by:

getPhase in interface org.springframework.context.Phased

handleMessageInternal

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

Deprecated.

Specified by:

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

Throws:

Exception

stop

public void stop()

Deprecated.

Specified by:

stop in interface org.springframework.context.Lifecycle