Package org.springframework.integration.aggregator

Class AggregatingMessageHandler

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.handler.MessageHandlerSupport

org.springframework.integration.handler.AbstractMessageHandler

org.springframework.integration.handler.AbstractMessageProducingHandler

org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler

org.springframework.integration.aggregator.AggregatingMessageHandler

All Implemented Interfaces:

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

public class AggregatingMessageHandler
extends AbstractCorrelatingMessageHandler

Aggregator specific implementation of AbstractCorrelatingMessageHandler. Will remove MessageGroups in the afterRelease(org.springframework.integration.store.MessageGroup, java.util.Collection<org.springframework.messaging.Message<?>>) only if 'expireGroupsUponCompletion' flag is set to 'true'.

Since:

2.1

Author:

Oleg Zhurakousky, Artem Bilan, Gary Russell, Ngoc Nhan

Nested Class Summary

Nested classes/interfaces inherited from class org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler

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

Method Summary

Modifier and Type	Method	Description
protected void	afterRelease(MessageGroup messageGroup, Collection<Message<?>> completedMessages)	Complete the group and remove all its messages.
IntegrationPatternType	getIntegrationPatternType()	Return a pattern type this component implements.
protected boolean	isExpireGroupsUponCompletion()
void	setExpireGroupsUponCompletion(boolean expireGroupsUponCompletion)	Will set the 'expireGroupsUponCompletion' flag.
protected boolean	shouldSplitOutput(Iterable<?> reply)	Check an Iterable result for split possibility on the output production: the items of the collection have to be instances of Message or AbstractIntegrationMessageBuilder and AbstractCorrelatingMessageHandler.getOutputProcessor() has to be a SimpleMessageGroupProcessor.

Methods inherited from class org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler

afterRelease, completeGroup, completeGroup, destroy, expireGroup, findLastReleasedSequenceNumber, forceComplete, getComponentType, getCorrelationStrategy, getDiscardChannel, getDiscardChannelName, getEvaluationContext, getExpireGroupScheduledFutures, getGroupConditionSupplier, getGroupTimeoutExpression, getLockRegistry, getMessageStore, getMinimumTimeoutForEmptyGroups, getOutputProcessor, getReleaseStrategy, handleMessageInternal, isLockRegistrySet, isReleaseLockBeforeSend, isReleasePartialSequences, isRunning, isSendPartialResultOnExpiry, isSequenceAware, obtainGroupTimeout, onInit, purgeOrphanedGroups, remove, setApplicationEventPublisher, setCorrelationStrategy, setDiscardChannel, setDiscardChannelName, setExpireDuration, setExpireDurationMillis, setExpireGroupsUponTimeout, setExpireTimeout, setForceReleaseAdviceChain, setGroupConditionSupplier, setGroupTimeoutExpression, setLockRegistry, setMessageStore, setMinimumTimeoutForEmptyGroups, setOutputProcessor, setPopSequence, setReleaseLockBeforeSend, setReleasePartialSequences, setReleaseStrategy, setSendPartialResultOnExpiry, start, stop, store, verifyResultCollectionConsistsOfMessages

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, setupMessageProcessor, shouldCopyRequestHeaders, updateNotPropagatedHeaders

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

handleMessage, onComplete, onError, onNext, onSubscribe, setObservationConvention

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

buildSendTimer, getManagedName, getManagedType, getMetricsCaptor, getObservationRegistry, getOrder, getOverrides, isLoggingEnabled, isObserved, registerMetricsCaptor, registerObservationRegistry, 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, getComponentDescription, getComponentName, getComponentSource, getConversionService, getExpression, getIntegrationProperties, getMessageBuilderFactory, getTaskScheduler, isInitialized, setApplicationContext, setBeanFactory, setBeanName, setChannelResolver, setComponentDescription, setComponentName, setComponentSource, 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

AggregatingMessageHandler

public AggregatingMessageHandler(MessageGroupProcessor processor,
 MessageGroupStore store,
 CorrelationStrategy correlationStrategy,
 ReleaseStrategy releaseStrategy)

AggregatingMessageHandler

public AggregatingMessageHandler(MessageGroupProcessor processor,
 MessageGroupStore store)

AggregatingMessageHandler

public AggregatingMessageHandler(MessageGroupProcessor processor)

Method Details

setExpireGroupsUponCompletion

public void setExpireGroupsUponCompletion(boolean expireGroupsUponCompletion)

Will set the 'expireGroupsUponCompletion' flag.

Parameters:

expireGroupsUponCompletion - true when groups should be expired on completion.

See Also:

• afterRelease(org.springframework.integration.store.MessageGroup, java.util.Collection<org.springframework.messaging.Message<?>>)

getIntegrationPatternType

public IntegrationPatternType getIntegrationPatternType()

Description copied from interface: IntegrationPattern

Return a pattern type this component implements.

Specified by:

getIntegrationPatternType in interface IntegrationPattern

Overrides:

getIntegrationPatternType in class MessageHandlerSupport

Returns:

the IntegrationPatternType this component implements.

isExpireGroupsUponCompletion

protected boolean isExpireGroupsUponCompletion()

Overrides:

isExpireGroupsUponCompletion in class AbstractCorrelatingMessageHandler

shouldSplitOutput

protected boolean shouldSplitOutput(Iterable<?> reply)

Check an Iterable result for split possibility on the output production: the items of the collection have to be instances of Message or AbstractIntegrationMessageBuilder and AbstractCorrelatingMessageHandler.getOutputProcessor() has to be a SimpleMessageGroupProcessor. Otherwise, a single reply message is emitted with the whole Iterable as its payload.

Overrides:

shouldSplitOutput in class AbstractMessageProducingHandler

Parameters:

reply - the Iterable result to check for split possibility.

Returns:

true if the Iterable result has to be split into individual messages.

Since:

6.0

afterRelease

protected void afterRelease(MessageGroup messageGroup,
 Collection<Message<?>> completedMessages)

Complete the group and remove all its messages. If the expireGroupsUponCompletion is true, then remove group fully.

Specified by:

afterRelease in class AbstractCorrelatingMessageHandler

Parameters:

messageGroup - the group to clean up.

completedMessages - The completed messages. Ignored in this implementation.