Package org.springframework.integration.dsl

Class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S,H>,H extends AbstractCorrelatingMessageHandler>

java.lang.Object

org.springframework.integration.dsl.IntegrationComponentSpec<S,reactor.util.function.Tuple2<F,H>>

org.springframework.integration.dsl.EndpointSpec<S,ConsumerEndpointFactoryBean,H>

org.springframework.integration.dsl.ConsumerEndpointSpec<S,H>

org.springframework.integration.dsl.CorrelationHandlerSpec<S,H>

Type Parameters:

S - the target CorrelationHandlerSpec implementation type.

H - the AbstractCorrelatingMessageHandler implementation type.

All Implemented Interfaces:

DisposableBean, FactoryBean<reactor.util.function.Tuple2<ConsumerEndpointFactoryBean,H>>, InitializingBean, Lifecycle, Phased, SmartLifecycle, ComponentsRegistration

Direct Known Subclasses:

AggregatorSpec, ResequencerSpec

public abstract class CorrelationHandlerSpec<S extends CorrelationHandlerSpec<S,H>,H extends AbstractCorrelatingMessageHandler>
extends ConsumerEndpointSpec<S,H>

A MessageHandlerSpec for an AbstractCorrelatingMessageHandler.

Since:

5.0

Author:

Artem Bilan, Gary Russell

Field Summary

Fields inherited from class org.springframework.integration.dsl.ConsumerEndpointSpec

adviceChain

Fields inherited from class org.springframework.integration.dsl.EndpointSpec

componentsToRegister, endpointFactoryBean, handler

Fields inherited from class org.springframework.integration.dsl.IntegrationComponentSpec

logger, PARSER, target

Fields inherited from interface org.springframework.beans.factory.FactoryBean

OBJECT_TYPE_ATTRIBUTE

Fields inherited from interface org.springframework.context.SmartLifecycle

DEFAULT_PHASE

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	CorrelationHandlerSpec(H messageHandler)

Method Summary

Modifier and Type	Method	Description
S	correlationExpression(String correlationExpression)	Configure the handler with an ExpressionEvaluatingCorrelationStrategy for the given expression.
S	correlationStrategy(Object target, String methodName)	Configure the handler with an MethodInvokingCorrelationStrategy for the target object and method name.
S	correlationStrategy(CorrelationStrategy correlationStrategy)
S	discardChannel(String discardChannelName)
S	discardChannel(MessageChannel discardChannel)
S	expireDuration(Duration expireDuration)	Configure a Duration how often to run a scheduled purge task.
S	expireGroupsUponTimeout(boolean expireGroupsUponTimeout)	Expire (completely remove) a group if it is completed due to timeout.
S	expireTimeout(long expireTimeout)	Configure a timeout for old groups in the store to purge.
S	forceReleaseAdvice(Advice... advice)	Configure a list of Advice objects to be applied to the forceComplete() operation.
S	groupConditionSupplier(BiFunction<Message<?>,String,String> conditionSupplier)	Configure a BiFunction to supply a group condition from a message to be added to the group.
S	groupTimeout(long groupTimeout)	Configure the handler with a group timeout expression that evaluates to this constant value.
S	groupTimeout(Function<MessageGroup,?> groupTimeoutFunction)	Configure the handler with a function that will be invoked to resolve the group timeout, based on the message group.
S	groupTimeoutExpression(String groupTimeoutExpression)	Specify a SpEL expression to evaluate the group timeout for scheduled expiration.
S	lockRegistry(LockRegistry lockRegistry)	Used to obtain a Lock based on the groupId for concurrent operations on the MessageGroup.
S	messageStore(MessageGroupStore messageStore)
S	minimumTimeoutForEmptyGroups(long minimumTimeoutForEmptyGroups)
S	popSequence(boolean popSequence)	Perform a MessageBuilder.popSequenceDetails() for output message or not.
S	processor(Object target)	Configure the handler with MethodInvokingCorrelationStrategy and MethodInvokingReleaseStrategy using the target object which should have methods annotated appropriately for each function.
S	releaseExpression(String releaseExpression)	Configure the handler with an ExpressionEvaluatingReleaseStrategy for the given expression.
S	releaseLockBeforeSend(boolean releaseLockBeforeSend)	Set to true to release the message group lock before sending any output.
S	releaseStrategy(Object target, String methodName)	Configure the handler with an MethodInvokingReleaseStrategy for the target object and method name.
S	releaseStrategy(ReleaseStrategy releaseStrategy)
S	sendPartialResultOnExpiry(boolean sendPartialResultOnExpiry)
S	taskScheduler(TaskScheduler taskScheduler)	Configure a TaskScheduler for scheduling tasks, for example in the Polling Consumer.

Methods inherited from class org.springframework.integration.dsl.ConsumerEndpointSpec

advice, async, autoStartup, customizeMonoReply, doGet, handleMessageAdvice, notPropagatedHeaders, order, phase, poller, reactive, reactive, requiresReply, role, sendTimeout, transactional, transactional, transactional, transactional, transactional

Methods inherited from class org.springframework.integration.dsl.EndpointSpec

assertHandler, getComponentsToRegister, id, obtainInputChannelFromFlow, obtainInputChannelFromFlow, poller, poller

Methods inherited from class org.springframework.integration.dsl.IntegrationComponentSpec

_this, afterPropertiesSet, destroy, getId, getObject, getObjectType, getPhase, isAutoStartup, isRunning, start, stop, stop

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.springframework.beans.factory.FactoryBean

isSingleton

Constructor Details

CorrelationHandlerSpec

protected CorrelationHandlerSpec(H messageHandler)

Method Details

messageStore

public S messageStore(MessageGroupStore messageStore)

Parameters:

messageStore - the message group store.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setMessageStore(MessageGroupStore)

sendPartialResultOnExpiry

public S sendPartialResultOnExpiry(boolean sendPartialResultOnExpiry)

Parameters:

sendPartialResultOnExpiry - the sendPartialResultOnExpiry.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setSendPartialResultOnExpiry(boolean)

minimumTimeoutForEmptyGroups

public S minimumTimeoutForEmptyGroups(long minimumTimeoutForEmptyGroups)

Parameters:

minimumTimeoutForEmptyGroups - the minimumTimeoutForEmptyGroups

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setMinimumTimeoutForEmptyGroups(long)

groupTimeout

public S groupTimeout(long groupTimeout)

Configure the handler with a group timeout expression that evaluates to this constant value.

Parameters:

groupTimeout - the group timeout in milliseconds.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setGroupTimeoutExpression(org.springframework.expression.Expression)
• ValueExpression

groupTimeoutExpression

public S groupTimeoutExpression(String groupTimeoutExpression)

Specify a SpEL expression to evaluate the group timeout for scheduled expiration. Must return Date, Long or String as a long.

Parameters:

groupTimeoutExpression - the group timeout expression string.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setGroupTimeoutExpression(org.springframework.expression.Expression)

groupTimeout

public S groupTimeout(Function<MessageGroup,?> groupTimeoutFunction)

Configure the handler with a function that will be invoked to resolve the group timeout, based on the message group. Usually used with a JDK8 lambda:

.groupTimeout(g -> g.size() * 2000L). Must return Date, Long or String a long.

Parameters:

groupTimeoutFunction - a function invoked to resolve the group timeout in milliseconds.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setGroupTimeoutExpression(org.springframework.expression.Expression)

taskScheduler

public S taskScheduler(TaskScheduler taskScheduler)

Description copied from class: ConsumerEndpointSpec

Configure a TaskScheduler for scheduling tasks, for example in the Polling Consumer. By default the global ThreadPoolTaskScheduler bean is used. This configuration is useful when there are requirements to dedicate particular threads for polling task, for example.

Overrides:

taskScheduler in class ConsumerEndpointSpec<S extends CorrelationHandlerSpec<S,H>,H extends AbstractCorrelatingMessageHandler>

Parameters:

taskScheduler - the task scheduler.

Returns:

the handler spec.

See Also:

• IntegrationObjectSupport.setTaskScheduler(TaskScheduler)

discardChannel

public S discardChannel(MessageChannel discardChannel)

Parameters:

discardChannel - the discard channel.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setDiscardChannel(MessageChannel)

discardChannel

public S discardChannel(String discardChannelName)

Parameters:

discardChannelName - the discard channel.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setDiscardChannelName(String)

processor

public S processor(Object target)

Configure the handler with MethodInvokingCorrelationStrategy and MethodInvokingReleaseStrategy using the target object which should have methods annotated appropriately for each function.

Parameters:

target - the target object

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setCorrelationStrategy(CorrelationStrategy)
• AbstractCorrelatingMessageHandler.setReleaseStrategy(ReleaseStrategy)

correlationExpression

public S correlationExpression(String correlationExpression)

Configure the handler with an ExpressionEvaluatingCorrelationStrategy for the given expression.

Parameters:

correlationExpression - the correlation expression.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setCorrelationStrategy(CorrelationStrategy)

correlationStrategy

public S correlationStrategy(Object target,
 String methodName)

Configure the handler with an MethodInvokingCorrelationStrategy for the target object and method name.

Parameters:

target - the target object.

methodName - the method name.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setCorrelationStrategy(CorrelationStrategy)

correlationStrategy

public S correlationStrategy(CorrelationStrategy correlationStrategy)

Parameters:

correlationStrategy - the correlation strategy.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setCorrelationStrategy(CorrelationStrategy)

releaseExpression

public S releaseExpression(String releaseExpression)

Configure the handler with an ExpressionEvaluatingReleaseStrategy for the given expression.

Parameters:

releaseExpression - the correlation expression.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setReleaseStrategy(ReleaseStrategy)

releaseStrategy

public S releaseStrategy(Object target,
 String methodName)

Configure the handler with an MethodInvokingReleaseStrategy for the target object and method name.

Parameters:

target - the target object.

methodName - the method name.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setReleaseStrategy(ReleaseStrategy)

releaseStrategy

public S releaseStrategy(ReleaseStrategy releaseStrategy)

Parameters:

releaseStrategy - the release strategy.

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setReleaseStrategy(ReleaseStrategy)

expireGroupsUponTimeout

public S expireGroupsUponTimeout(boolean expireGroupsUponTimeout)

Expire (completely remove) a group if it is completed due to timeout. Default true for aggregator and false for resequencer.

Parameters:

expireGroupsUponTimeout - the expireGroupsUponTimeout to set

Returns:

the handler spec.

See Also:

• AbstractCorrelatingMessageHandler.setExpireGroupsUponTimeout(boolean)

forceReleaseAdvice

public S forceReleaseAdvice(Advice... advice)

Configure a list of Advice objects to be applied to the forceComplete() operation.

Parameters:

advice - the advice chain.

Returns:

the endpoint spec.

lockRegistry

public S lockRegistry(LockRegistry lockRegistry)

Used to obtain a Lock based on the groupId for concurrent operations on the MessageGroup. By default, an internal DefaultLockRegistry is used. Use of a distributed LockRegistry, such as the RedisLockRegistry, ensures only one instance of the aggregator will operate on a group concurrently.

Parameters:

lockRegistry - the LockRegistry to use.

Returns:

the endpoint spec.

popSequence

public S popSequence(boolean popSequence)

Perform a MessageBuilder.popSequenceDetails() for output message or not.

Parameters:

popSequence - the boolean flag to use.

Returns:

the endpoint spec.

Since:

5.1

See Also:

• AbstractCorrelatingMessageHandler.setPopSequence(boolean)

expireTimeout

public S expireTimeout(long expireTimeout)

Configure a timeout for old groups in the store to purge.

Parameters:

expireTimeout - the timeout in milliseconds to use.

Returns:

the endpoint spec.

Since:

5.5

See Also:

• AbstractCorrelatingMessageHandler.setExpireTimeout(long)

expireDuration

public S expireDuration(Duration expireDuration)

Configure a Duration how often to run a scheduled purge task.

Parameters:

expireDuration - the duration for scheduled purge task.

Returns:

the endpoint spec.

Since:

5.5

See Also:

• AbstractCorrelatingMessageHandler.setExpireDuration(Duration)

releaseLockBeforeSend

public S releaseLockBeforeSend(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.

Returns:

the endpoint spec.

Since:

5.5

See Also:

• AbstractCorrelatingMessageHandler.setReleaseLockBeforeSend(boolean)

groupConditionSupplier

public S groupConditionSupplier(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.

Returns:

the endpoint spec.

Since:

5.5