org.springframework.integration.aggregator

Class CorrelatingMessageBarrier

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.handler.AbstractMessageHandler

org.springframework.integration.aggregator.CorrelatingMessageBarrier

All Implemented Interfaces:

org.reactivestreams.Subscriber<org.springframework.messaging.Message<?>>, 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, ExpressionCapable, Orderable, MessageSource<java.lang.Object>, NamedComponent, ConfigurableMetricsAware<AbstractMessageHandlerMetrics>, IntegrationManagement, IntegrationStatsManagement, MessageHandlerMetrics, TrackableComponent, org.springframework.messaging.MessageHandler, reactor.core.CoreSubscriber<org.springframework.messaging.Message<?>>

public class CorrelatingMessageBarrier
extends AbstractMessageHandler
implements MessageSource<java.lang.Object>

This Endpoint serves as a barrier for messages that should not be processed yet. The decision when a message can be processed is delegated to a ReleaseStrategy. When a message can be processed it is up to the client to take care of the locking (potentially from the ReleaseStrategy's canRelease(..) method).

This class differs from AbstractCorrelatingMessageHandler in that it completely decouples the receiver and the sender. It can be applied in scenarios where completion of a message group is not well defined but only a certain amount of messages for any given correlation key may be processed at a time.

The messages will be stored in a MessageStore for each correlation key.

Author:

Iwein Fuld, Oleg Zhurakousky, Gary Russell

See Also:

AbstractCorrelatingMessageHandler

Nested Class Summary

Nested classes/interfaces inherited from interface org.springframework.integration.support.management.IntegrationManagement

IntegrationManagement.ManagementOverrides

Field Summary

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

EXPRESSION_PARSER, logger

Fields inherited from interface org.springframework.integration.support.management.IntegrationManagement

METER_PREFIX, RECEIVE_COUNTER_NAME, SEND_TIMER_NAME

Fields inherited from interface org.springframework.core.Ordered

HIGHEST_PRECEDENCE, LOWEST_PRECEDENCE

Constructor Summary

Constructors

Constructor and Description
CorrelatingMessageBarrier()
CorrelatingMessageBarrier(MessageGroupStore store)

Method Summary

Modifier and Type	Method and Description
protected void	handleMessageInternal(org.springframework.messaging.Message<?> message)
org.springframework.messaging.Message<java.lang.Object>	receive() Retrieve the next available message from this source.
void	setCorrelationStrategy(CorrelationStrategy correlationStrategy) Set the CorrelationStrategy to be used to determine the correlation key for incoming messages
void	setReleaseStrategy(ReleaseStrategy releaseStrategy) Set the ReleaseStrategy that should be used when deciding if a group in this barrier may be released.

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

configureMetrics, destroy, getActiveCount, getActiveCountLong, getComponentType, getDuration, getErrorCount, getErrorCountLong, getHandleCount, getHandleCountLong, getManagedName, getManagedType, getMaxDuration, getMeanDuration, getMinDuration, getOrder, getOverrides, getStandardDeviationDuration, handleMessage, isCountsEnabled, isLoggingEnabled, isStatsEnabled, onComplete, onError, onInit, onNext, onSubscribe, registerMetricsCaptor, reset, setCountsEnabled, setLoggingEnabled, setManagedName, setManagedType, setOrder, setShouldTrack, setStatsEnabled

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

afterPropertiesSet, extractTypeIfPossible, generateId, getApplicationContext, getApplicationContextId, getBeanFactory, getChannelResolver, getComponentName, getConversionService, getExpression, getIntegrationProperties, getIntegrationProperty, getMessageBuilderFactory, getTaskScheduler, isInitialized, 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

Methods inherited from interface reactor.core.CoreSubscriber

currentContext

Constructor Detail

CorrelatingMessageBarrier

public CorrelatingMessageBarrier(MessageGroupStore store)

CorrelatingMessageBarrier

public CorrelatingMessageBarrier()

Method Detail

setCorrelationStrategy

public void setCorrelationStrategy(CorrelationStrategy correlationStrategy)

Set the CorrelationStrategy to be used to determine the correlation key for incoming messages

Parameters:

correlationStrategy - The correlation strategy.

setReleaseStrategy

public void setReleaseStrategy(ReleaseStrategy releaseStrategy)

Set the ReleaseStrategy that should be used when deciding if a group in this barrier may be released.

Parameters:

releaseStrategy - The release strategy.

handleMessageInternal

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

Specified by:

handleMessageInternal in class AbstractMessageHandler

Throws:

java.lang.Exception

receive

public org.springframework.messaging.Message<java.lang.Object> receive()

Description copied from interface: MessageSource

Retrieve the next available message from this source. Returns null if no message is available.

Specified by:

receive in interface MessageSource<java.lang.Object>

Returns:

The message or null.