Package org.springframework.integration.aggregator

Class FluxAggregatorMessageHandler

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.FluxAggregatorMessageHandler

All Implemented Interfaces:

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

public class FluxAggregatorMessageHandler
extends AbstractMessageProducingHandler
implements ManageableLifecycle

The AbstractMessageProducingHandler implementation for aggregation logic based on Reactor's Flux.groupBy(java.util.function.Function<? super T, ? extends K>) and Flux.window(int) operators.

The incoming messages are emitted into a FluxSink provided by the Flux.create(java.util.function.Consumer<? super reactor.core.publisher.FluxSink<T>>) initialized in the constructor.

The resulting windows for groups are wrapped into Messages for downstream consumption.

If the AbstractMessageProducingHandler.getOutputChannel() is not a ReactiveStreamsSubscribableChannel instance, a subscription for the whole aggregating Flux is performed in the start() method.

Since:

5.2

Author:

Artem Bilan

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.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
FluxAggregatorMessageHandler()	Create an instance with a Flux.create(java.util.function.Consumer<? super reactor.core.publisher.FluxSink<T>>) and apply Flux.groupBy(java.util.function.Function<? super T, ? extends K>) and Flux.window(int) transformation into it.

Method Summary

Modifier and Type	Method	Description
String	getComponentType()	Subclasses may implement this method to provide component type information.
IntegrationPatternType	getIntegrationPatternType()	Return a pattern type this component implements.
protected void	handleMessageInternal(Message<?> message)
boolean	isRunning()
void	setBoundaryTrigger(Predicate<Message<?>> boundaryTrigger)	Configure a Predicate for messages to determine a window boundary in the Flux.windowUntil(java.util.function.Predicate<T>) operator.
void	setCombineFunction(Function<reactor.core.publisher.Flux<Message<?>>,reactor.core.publisher.Mono<Message<?>>> combineFunction)	Configure a transformation Function to apply for a Flux window to emit.
void	setCorrelationStrategy(CorrelationStrategy correlationStrategy)	Configure a CorrelationStrategy to determine a group key from the incoming messages.
void	setWindowConfigurer(Function<reactor.core.publisher.Flux<Message<?>>,reactor.core.publisher.Flux<reactor.core.publisher.Flux<Message<?>>>> windowConfigurer)	Configure a Function to apply a transformation into the grouping Flux for any arbitrary Flux.window(int) options not covered by the simple options.
void	setWindowSize(int windowSize)	Specify a size for windows to close.
void	setWindowSizeFunction(Function<Message<?>,Integer> windowSizeFunction)	Specify a Function to determine a size for windows to close against the first message in group.
void	setWindowTimespan(Duration windowTimespan)	Configure a Duration for closing windows periodically.
protected boolean	shouldCopyRequestHeaders()	Subclasses may override this.
void	start()
void	stop()

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

addNotPropagatedHeaders, createOutputMessage, getNotPropagatedHeaders, getOutputChannel, isAsync, messageBuilderForReply, onInit, produceOutput, resolveErrorChannel, sendErrorMessage, sendOutput, sendOutputs, setAsync, setNotPropagatedHeaders, setOutputChannel, setOutputChannelName, setSendTimeout, setupMessageProcessor, shouldSplitOutput, 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, destroy, 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, 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.integration.support.context.NamedComponent

getBeanName, getComponentName

Constructor Details

FluxAggregatorMessageHandler

public FluxAggregatorMessageHandler()

Create an instance with a Flux.create(java.util.function.Consumer<? super reactor.core.publisher.FluxSink<T>>) and apply Flux.groupBy(java.util.function.Function<? super T, ? extends K>) and Flux.window(int) transformation into it.

Method Details

setCorrelationStrategy

public void setCorrelationStrategy(CorrelationStrategy correlationStrategy)

Configure a CorrelationStrategy to determine a group key from the incoming messages. By default a HeaderAttributeCorrelationStrategy is used against a IntegrationMessageHeaderAccessor.CORRELATION_ID header value.

Parameters:

correlationStrategy - the CorrelationStrategy to use.

setCombineFunction

public void setCombineFunction(Function<reactor.core.publisher.Flux<Message<?>>,reactor.core.publisher.Mono<Message<?>>> combineFunction)

Configure a transformation Function to apply for a Flux window to emit. Requires a Mono result with a Message as value as a combination result of the incoming Flux for window. By default a Flux for window is fully wrapped into a message with headers copied from the first message in window. Such a Flux in the payload has to be subscribed and consumed downstream.

Parameters:

combineFunction - the Function to use for result windows transformation.

setBoundaryTrigger

public void setBoundaryTrigger(Predicate<Message<?>> boundaryTrigger)

Configure a Predicate for messages to determine a window boundary in the Flux.windowUntil(java.util.function.Predicate<T>) operator. Has a precedence over any other window configuration options.

Parameters:

boundaryTrigger - the Predicate to use for window boundary.

See Also:

• Flux.windowUntil(Predicate)

setWindowSize

public void setWindowSize(int windowSize)

Specify a size for windows to close. Can be combined with the setWindowTimespan(Duration).

Parameters:

windowSize - the size for window to use.

See Also:

• Flux.window(int)
• Flux.windowTimeout(int, Duration)

setWindowSizeFunction

public void setWindowSizeFunction(Function<Message<?>,Integer> windowSizeFunction)

Specify a Function to determine a size for windows to close against the first message in group. Tne result of the function can be combined with the setWindowTimespan(Duration). By default an IntegrationMessageHeaderAccessor.SEQUENCE_SIZE header is consulted.

Parameters:

windowSizeFunction - the Function to use to determine a window size against a first message in the group.

See Also:

• Flux.window(int)
• Flux.windowTimeout(int, Duration)

setWindowTimespan

public void setWindowTimespan(Duration windowTimespan)

Configure a Duration for closing windows periodically. Can be combined with the setWindowSize(int) or setWindowSizeFunction(Function).

Parameters:

windowTimespan - the Duration to use for windows to close periodically.

See Also:

• Flux.window(Duration)
• Flux.windowTimeout(int, Duration)

setWindowConfigurer

public void setWindowConfigurer(Function<reactor.core.publisher.Flux<Message<?>>,reactor.core.publisher.Flux<reactor.core.publisher.Flux<Message<?>>>> windowConfigurer)

Configure a Function to apply a transformation into the grouping Flux for any arbitrary Flux.window(int) options not covered by the simple options. Has a precedence over any other window configuration options.

Parameters:

windowConfigurer - the Function to apply any custom window transformation.

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

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.

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

handleMessageInternal

protected void handleMessageInternal(Message<?> message)

Specified by:

handleMessageInternal in class AbstractMessageHandler

shouldCopyRequestHeaders

protected boolean shouldCopyRequestHeaders()

Description copied from class: AbstractMessageProducingHandler

Subclasses may override this. True by default.

Overrides:

shouldCopyRequestHeaders in class AbstractMessageProducingHandler

Returns:

true if the request headers should be copied.