org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler

All Implemented Interfaces:: org.reactivestreams.Subscriber<Message<?>>, Aware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, ApplicationContextAware, ApplicationEventPublisherAware, Lifecycle, Ordered, ExpressionCapable, Orderable, MessageProducer, DiscardingMessageHandler, HeaderPropagationAware, IntegrationPattern, NamedComponent, IntegrationManagement, ManageableLifecycle, TrackableComponent, MessageHandler, reactor.core.CoreSubscriber<Message<?>>

Direct Known Subclasses:: AggregatingMessageHandler, ResequencingMessageHandler

public abstract class AbstractCorrelatingMessageHandler
extends AbstractMessageProducingHandler
implements DiscardingMessageHandler, ApplicationEventPublisherAware, ManageableLifecycle

Abstract Message handler that holds a buffer of correlated messages in a MessageStore. This class takes care of correlated groups of messages that can be completed in batches. It is useful for custom implementation of MessageHandlers that require correlation and is used as a base class for Aggregator - AggregatingMessageHandler and Resequencer - ResequencingMessageHandler, or custom implementations requiring correlation.

To customize this handler inject CorrelationStrategy, ReleaseStrategy, and MessageGroupProcessor implementations as you require.

By default the CorrelationStrategy will be a HeaderAttributeCorrelationStrategy and the ReleaseStrategy will be a SequenceSizeReleaseStrategy.

Use proper CorrelationStrategy for cases when same MessageStore is used for multiple handlers to ensure uniqueness of message groups across handlers.

When the expireTimeout is greater than 0, groups which are older than this timeout are purged from the store on start up (or when purgeOrphanedGroups() is called). If expireDuration is provided, the task is scheduled to perform purgeOrphanedGroups() periodically.

Since:: 2.0
Author:: Iwein Fuld, Dave Syer, Oleg Zhurakousky, Gary Russell, Artem Bilan, David Liu, Enrique Rodriguez, Meherzad Lahewala, Jayadev Sirimamilla

Nested Class Summary

Nested Classes

Modifier and Type Class Description

protected static class AbstractCorrelatingMessageHandler.SequenceAwareMessageGroup

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

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

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	Description
`AbstractCorrelatingMessageHandler(MessageGroupProcessor processor)`
`AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store)`
`AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy)`

Method Summary

Modifier and Type	Method	Description
`protected abstract void`	`afterRelease(MessageGroup group, Collection<Message<?>> completedMessages)`	Allows you to provide additional logic that needs to be performed after the MessageGroup was released.
`protected void`	`afterRelease(MessageGroup group, Collection<Message<?>> completedMessages, boolean timeout)`	Subclasses may override if special action is needed because the group was released or discarded due to a timeout.
`protected void`	`completeGroup(Object correlationKey, MessageGroup group, Lock lock)`
`protected Collection<Message<?>>`	`completeGroup(Message<?> message, Object correlationKey, MessageGroup group, Lock lock)`
`void`	`destroy()`
`protected void`	`expireGroup(Object correlationKey, MessageGroup group, Lock lock)`
`protected int`	`findLastReleasedSequenceNumber(Object groupId, Collection<Message<?>> partialSequence)`
`protected void`	`forceComplete(MessageGroup group)`
`String`	`getComponentType()`	Subclasses may implement this method to provide component type information.
`protected CorrelationStrategy`	`getCorrelationStrategy()`
`MessageChannel`	`getDiscardChannel()`	Return the discard channel.
`protected String`	`getDiscardChannelName()`
`protected EvaluationContext`	`getEvaluationContext()`
`protected Map<UUID,ScheduledFuture<?>>`	`getExpireGroupScheduledFutures()`
`protected BiFunction<Message<?>,String,String>`	`getGroupConditionSupplier()`
`protected Expression`	`getGroupTimeoutExpression()`
`protected LockRegistry`	`getLockRegistry()`
`MessageGroupStore`	`getMessageStore()`
`protected long`	`getMinimumTimeoutForEmptyGroups()`
`MessageGroupProcessor`	`getOutputProcessor()`	Return a configured `MessageGroupProcessor`.
`protected ReleaseStrategy`	`getReleaseStrategy()`
`protected void`	`handleMessageInternal(Message<?> message)`
`protected boolean`	`isExpireGroupsUponCompletion()`
`protected boolean`	`isLockRegistrySet()`
`protected boolean`	`isReleaseLockBeforeSend()`
`protected boolean`	`isReleasePartialSequences()`
`boolean`	`isRunning()`
`protected boolean`	`isSendPartialResultOnExpiry()`
`protected boolean`	`isSequenceAware()`
`protected Object`	`obtainGroupTimeout(MessageGroup group)`
`protected void`	`onInit()`	Subclasses may implement this for initialization logic.
`void`	`purgeOrphanedGroups()`	Perform a `MessageGroupStore.expireMessageGroups(long)` with the provided `expireTimeout`.
`protected void`	`remove(MessageGroup group)`
`void`	`setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher)`
`void`	`setCorrelationStrategy(CorrelationStrategy correlationStrategy)`
`void`	`setDiscardChannel(MessageChannel discardChannel)`
`void`	`setDiscardChannelName(String discardChannelName)`
`void`	`setExpireDuration(Duration expireDuration)`	Configure a `Duration` how often to clean up old orphaned groups from the store.
`void`	`setExpireDurationMillis(long expireDuration)`	Configure a `Duration` (in millis) how often to clean up old orphaned groups from the store.
`void`	`setExpireGroupsUponTimeout(boolean expireGroupsUponTimeout)`	Expire (completely remove) a group if it is completed due to timeout.
`void`	`setExpireTimeout(long expireTimeout)`	Configure a timeout in milliseconds for purging old orphaned groups from the store.
`void`	`setForceReleaseAdviceChain(List<Advice> forceReleaseAdviceChain)`
`void`	`setGroupConditionSupplier(BiFunction<Message<?>,String,String> conditionSupplier)`	Configure a `BiFunction` to supply a group condition from a message to be added to the group.
`void`	`setGroupTimeoutExpression(Expression groupTimeoutExpression)`
`void`	`setLockRegistry(LockRegistry lockRegistry)`
`void`	`setMessageStore(MessageGroupStore store)`
`void`	`setMinimumTimeoutForEmptyGroups(long minimumTimeoutForEmptyGroups)`	By default, when a MessageGroupStoreReaper is configured to expire partial groups, empty groups are also removed.
`void`	`setOutputProcessor(MessageGroupProcessor outputProcessor)`	Specify a `MessageGroupProcessor` for the output function.
`void`	`setPopSequence(boolean popSequence)`	Perform a `MessageBuilder.popSequenceDetails()` for output message or not.
`void`	`setReleaseLockBeforeSend(boolean releaseLockBeforeSend)`	Set to true to release the message group lock before sending any output.
`void`	`setReleasePartialSequences(boolean releasePartialSequences)`	Set `releasePartialSequences` on an underlying default `SequenceSizeReleaseStrategy`.
`void`	`setReleaseStrategy(ReleaseStrategy releaseStrategy)`
`void`	`setSendPartialResultOnExpiry(boolean sendPartialResultOnExpiry)`
`void`	`start()`
`void`	`stop()`
`protected MessageGroup`	`store(Object correlationKey, Message<?> message)`
`protected void`	`verifyResultCollectionConsistsOfMessages(Collection<?> elements)`

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

addNotPropagatedHeaders, createOutputMessage, getNotPropagatedHeaders, getOutputChannel, isAsync, messageBuilderForReply, produceOutput, resolveErrorChannel, sendErrorMessage, sendOutput, sendOutputs, setAsync, setNotPropagatedHeaders, setOutputChannel, setOutputChannelName, setSendTimeout, shouldCopyRequestHeaders, shouldSplitOutput, updateNotPropagatedHeaders

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

handleMessage, onComplete, onError, onNext, onSubscribe

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

buildSendTimer, getIntegrationPatternType, getManagedName, getManagedType, getMetricsCaptor, getOrder, getOverrides, isLoggingEnabled, registerMetricsCaptor, sendTimer, setLoggingEnabled, setManagedName, setManagedType, setOrder, setShouldTrack, shouldTrack

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

afterPropertiesSet, extractTypeIfPossible, generateId, getApplicationContext, getApplicationContextId, getBeanDescription, getBeanFactory, getBeanName, 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 reactor.core.CoreSubscriber

currentContext

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

getThisAs

Methods inherited from interface org.springframework.messaging.MessageHandler

handleMessage

Methods inherited from interface org.springframework.integration.support.context.NamedComponent

getBeanName, getComponentName

Constructor Details
- AbstractCorrelatingMessageHandler
  
  public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy)
- AbstractCorrelatingMessageHandler
  
  public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store)
- AbstractCorrelatingMessageHandler
  
  public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor)
Method Details
- setLockRegistry
  
  public void setLockRegistry(LockRegistry lockRegistry)
- setMessageStore
  
  public final void setMessageStore(MessageGroupStore store)
- setCorrelationStrategy
  
  public void setCorrelationStrategy(CorrelationStrategy correlationStrategy)
- setReleaseStrategy
  
  public void setReleaseStrategy(ReleaseStrategy releaseStrategy)
- setGroupTimeoutExpression
  
  public void setGroupTimeoutExpression(Expression groupTimeoutExpression)
- setForceReleaseAdviceChain
  
  public void setForceReleaseAdviceChain(List<Advice> forceReleaseAdviceChain)
- setOutputProcessor
  
  public void setOutputProcessor(MessageGroupProcessor outputProcessor)
  
  Specify a MessageGroupProcessor for the output function.
  
  Parameters:
  
  outputProcessor - the MessageGroupProcessor to use
  
  Since:
  
  5.0
- getOutputProcessor
  
  public MessageGroupProcessor getOutputProcessor()
  
  Return a configured MessageGroupProcessor.
  
  Returns:
  
  the configured MessageGroupProcessor
  
  Since:
  
  5.2
- setDiscardChannel
  
  public void setDiscardChannel(MessageChannel discardChannel)
- setDiscardChannelName
  
  public void setDiscardChannelName(String discardChannelName)
- setSendPartialResultOnExpiry
  
  public void setSendPartialResultOnExpiry(boolean sendPartialResultOnExpiry)
- setMinimumTimeoutForEmptyGroups
  
  public void setMinimumTimeoutForEmptyGroups(long minimumTimeoutForEmptyGroups)
  
  By default, when a MessageGroupStoreReaper is configured to expire partial groups, empty groups are also removed. Empty groups exist after a group is released normally. This is to enable the detection and discarding of late-arriving messages. If you wish to expire empty groups on a longer schedule than expiring partial groups, set this property. Empty groups will then not be removed from the MessageStore until they have not been modified for at least this number of milliseconds.
  
  Parameters:
  
  minimumTimeoutForEmptyGroups - The minimum timeout.
- setReleasePartialSequences
  
  public void setReleasePartialSequences(boolean releasePartialSequences)
  
  Set releasePartialSequences on an underlying default SequenceSizeReleaseStrategy. Ignored for other release strategies.
  
  Parameters:
  
  releasePartialSequences - true to allow release.
- setExpireGroupsUponTimeout
  
  public void setExpireGroupsUponTimeout(boolean expireGroupsUponTimeout)
  
  Expire (completely remove) a group if it is completed due to timeout. Default true
  
  Parameters:
  
  expireGroupsUponTimeout - the expireGroupsUponTimeout to set
  
  Since:
  
  4.1
- setPopSequence
  
  public void setPopSequence(boolean popSequence)
  
  Perform a MessageBuilder.popSequenceDetails() for output message or not. Default to true. This option removes the sequence information added by the nearest upstream component with applySequence=true (for example splitter).
  
  Parameters:
  
  popSequence - the boolean flag to use.
  
  Since:
  
  5.1
- isReleaseLockBeforeSend
  
  protected boolean isReleaseLockBeforeSend()
- setReleaseLockBeforeSend
  
  public void setReleaseLockBeforeSend(boolean releaseLockBeforeSend)
  
  Set to true to release the message group lock before sending any output. See "Avoiding Deadlocks" in the Aggregator section of the reference manual for more information as to why this might be needed.
  
  Parameters:
  
  releaseLockBeforeSend - true to release the lock.
  
  Since:
  
  5.1.1
- setExpireTimeout
  
  public void setExpireTimeout(long expireTimeout)
  
  Configure a timeout in milliseconds for purging old orphaned groups from the store. Used on startup and when an expireDuration is provided, the task for running purgeOrphanedGroups() is scheduled with that period. The forceReleaseProcessor is used to process those expired groups according the "force complete" options. A group can be orphaned if a persistent message group store is used and no new messages arrive for that group after a restart.
  
  Parameters:
  
  expireTimeout - the number of milliseconds to determine old orphaned groups in the store to purge.
  
  Since:
  
  5.4
  
  See Also:
  
  purgeOrphanedGroups()
- setExpireDurationMillis
  
  public void setExpireDurationMillis(long expireDuration)
  
  Configure a Duration (in millis) how often to clean up old orphaned groups from the store.
  
  Parameters:
  
  expireDuration - the delay how often to call purgeOrphanedGroups().
  
  Since:
  
  5.4
  
  See Also:
  
  purgeOrphanedGroups(), setExpireDuration(Duration), setExpireTimeout(long)
- setExpireDuration
  
  public void setExpireDuration(@Nullable Duration expireDuration)
  
  Configure a Duration how often to clean up old orphaned groups from the store.
  
  Parameters:
  
  expireDuration - the delay how often to call purgeOrphanedGroups().
  
  Since:
  
  5.4
  
  See Also:
  
  purgeOrphanedGroups(), setExpireTimeout(long)
- setGroupConditionSupplier
  
  public void setGroupConditionSupplier(BiFunction<Message<?>,String,String> conditionSupplier)
  
  Configure a BiFunction to supply a group condition from a message to be added to the group. The null result from the function will reset a condition set before.
  
  Parameters:
  
  conditionSupplier - the function to supply a group condition from a message to be added to the group.
  
  Since:
  
  5.5
  
  See Also:
  
  GroupConditionProvider
- setApplicationEventPublisher
  
  public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher)
  
  Specified by:
  
  setApplicationEventPublisher in interface ApplicationEventPublisherAware
- onInit
  
  protected void onInit()
  
  Description copied from class: IntegrationObjectSupport
  
  Subclasses may implement this for initialization logic.
  
  Overrides:
  
  onInit in class AbstractMessageProducingHandler
- getComponentType
  
  public String getComponentType()
  
  Description copied from class: IntegrationObjectSupport
  
  Subclasses may implement this method to provide component type information.
  
  Specified by:
  
  getComponentType in interface NamedComponent
  
  Overrides:
  
  getComponentType in class MessageHandlerSupport
- getMessageStore
  
  public MessageGroupStore getMessageStore()
- getExpireGroupScheduledFutures
  
  protected Map<UUID,ScheduledFuture<?>> getExpireGroupScheduledFutures()
- getCorrelationStrategy
  
  protected CorrelationStrategy getCorrelationStrategy()
- getReleaseStrategy
  
  protected ReleaseStrategy getReleaseStrategy()
- getGroupConditionSupplier
  
  @Nullable protected BiFunction<Message<?>,String,String> getGroupConditionSupplier()
- getDiscardChannel
  
  public MessageChannel getDiscardChannel()
  
  Description copied from interface: DiscardingMessageHandler
  
  Return the discard channel.
  
  Specified by:
  
  getDiscardChannel in interface DiscardingMessageHandler
  
  Returns:
  
  the channel.
- getDiscardChannelName
  
  protected String getDiscardChannelName()
- isSendPartialResultOnExpiry
  
  protected boolean isSendPartialResultOnExpiry()
- isSequenceAware
  
  protected boolean isSequenceAware()
- getLockRegistry
  
  protected LockRegistry getLockRegistry()
- isLockRegistrySet
  
  protected boolean isLockRegistrySet()
- getMinimumTimeoutForEmptyGroups
  
  protected long getMinimumTimeoutForEmptyGroups()
- isReleasePartialSequences
  
  protected boolean isReleasePartialSequences()
- getGroupTimeoutExpression
  
  protected Expression getGroupTimeoutExpression()
- getEvaluationContext
  
  protected EvaluationContext getEvaluationContext()
- handleMessageInternal
  
  protected void handleMessageInternal(Message<?> message)
  
  Specified by:
  
  handleMessageInternal in class AbstractMessageHandler
- isExpireGroupsUponCompletion
  
  protected boolean isExpireGroupsUponCompletion()
- afterRelease
  
  protected abstract void afterRelease(MessageGroup group, Collection<Message<?>> completedMessages)
  
  Allows you to provide additional logic that needs to be performed after the MessageGroup was released.
  
  Parameters:
  
  group - The group.
  
  completedMessages - The completed messages.
- afterRelease
  
  protected void afterRelease(MessageGroup group, Collection<Message<?>> completedMessages, boolean timeout)
  
  Subclasses may override if special action is needed because the group was released or discarded due to a timeout. By default, afterRelease(MessageGroup, Collection) is invoked.
  
  Parameters:
  
  group - The group.
  
  completedMessages - The completed messages.
  
  timeout - True if the release/discard was due to a timeout.
- forceComplete
  
  protected void forceComplete(MessageGroup group)
- remove
  
  protected void remove(MessageGroup group)
- findLastReleasedSequenceNumber
  
  protected int findLastReleasedSequenceNumber(Object groupId, Collection<Message<?>> partialSequence)
- store
  
  protected MessageGroup store(Object correlationKey, Message<?> message)
- expireGroup
  
  protected void expireGroup(Object correlationKey, MessageGroup group, Lock lock)
- completeGroup
  
  protected void completeGroup(Object correlationKey, MessageGroup group, Lock lock)
- completeGroup
  
  protected Collection<Message<?>> completeGroup(Message<?> message, Object correlationKey, MessageGroup group, Lock lock)
- verifyResultCollectionConsistsOfMessages
  
  protected void verifyResultCollectionConsistsOfMessages(Collection<?> elements)
- obtainGroupTimeout
  
  protected Object obtainGroupTimeout(MessageGroup group)
- destroy
  
  public void destroy()
  
  Specified by:
  
  destroy in interface DisposableBean
  
  Specified by:
  
  destroy in interface IntegrationManagement
  
  Overrides:
  
  destroy in class MessageHandlerSupport
- start
  
  public void start()
  
  Specified by:
  
  start in interface Lifecycle
  
  Specified by:
  
  start in interface ManageableLifecycle
- stop
  
  public void stop()
  
  Specified by:
  
  stop in interface Lifecycle
  
  Specified by:
  
  stop in interface ManageableLifecycle
- isRunning
  
  public boolean isRunning()
  
  Specified by:
  
  isRunning in interface Lifecycle
  
  Specified by:
  
  isRunning in interface ManageableLifecycle
- purgeOrphanedGroups
  
  public void purgeOrphanedGroups()
  
  Perform a MessageGroupStore.expireMessageGroups(long) with the provided expireTimeout. Can be called externally at any time. Internally it is called from the scheduled task with the configured expireDuration.
  
  Since:
  
  5.4

Class AbstractCorrelatingMessageHandler

Nested Class Summary

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

Field Summary

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

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

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

Fields inherited from interface org.springframework.core.Ordered