org.springframework.xd.rxjava

Class MultipleSubjectMessageHandler

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.handler.AbstractMessageHandler

org.springframework.integration.handler.AbstractMessageProducingHandler

org.springframework.xd.rxjava.MultipleSubjectMessageHandler

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

public class MultipleSubjectMessageHandler
extends org.springframework.integration.handler.AbstractMessageProducingHandler
implements org.springframework.beans.factory.DisposableBean, org.springframework.integration.expression.IntegrationEvaluationContextAware

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

The specific Observable that the message is delegated 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 observable. This is due to the underlying use of a PublishSubject.

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

If the Observable 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

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
MultipleSubjectMessageHandler(Processor processor, java.lang.String partitionExpression)

Method Summary

Methods

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

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

MultipleSubjectMessageHandler

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

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

setIntegrationEvaluationContext

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

Specified by:

setIntegrationEvaluationContext in interface org.springframework.integration.expression.IntegrationEvaluationContextAware