org.axonframework.config

Class EventHandlingConfiguration

java.lang.Object

org.axonframework.config.EventHandlingConfiguration

All Implemented Interfaces:

ModuleConfiguration

public class EventHandlingConfiguration
extends Object
implements ModuleConfiguration

Module Configuration implementation that defines an Event Handling component. Typically, such a configuration consists of a number of Event Handlers and one or more Event Processors that define the transactional semantics of the processing. Each Event Handler is assigned to one Event Processor.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static interface	EventHandlingConfiguration.EventProcessorBuilder Interface describing a Builder function for Event Processors.

Constructor Summary

Constructors

Constructor and Description
EventHandlingConfiguration() Creates a default configuration for an Event Handling module that creates a SubscribingEventProcessor instance for all Event Handlers that have the same Processing Group name.

Method Summary

Modifier and Type	Method and Description
EventHandlingConfiguration	assignHandlersMatching(String name, int priority, Predicate<Object> criteria) Configures a rule to assign Event Handler beans that match the given criteria to the Event Processor with given name, with given priority.
EventHandlingConfiguration	assignHandlersMatching(String name, Predicate<Object> criteria) Configures a rule to assign Event Handler beans that match the given criteria to the Event Processor with given name, with neutral priority (value 0).
EventHandlingConfiguration	byDefaultAssignTo(Function<Object,String> assignmentFunction) Registers a function that defines the Event Processor name to assign Event Handler beans to when no other, more explicit, rule matches and no ProcessingGroup annotation is found.
EventHandlingConfiguration	byDefaultAssignTo(String name) Registers the Event Processor name to assign Event Handler beans to when no other, more explicit, rule matches and no ProcessingGroup annotation is found.
EventHandlingConfiguration	configureMessageMonitor(String name, Function<Configuration,MessageMonitor<Message<?>>> messageMonitorBuilder) Configures the builder function to create the Message Monitor for the EventProcessor of the given name.
EventHandlingConfiguration	configureMessageMonitor(String name, MessageMonitorFactory messageMonitorFactory) Configures the factory to create the Message Monitor for the EventProcessor of the given name.
<T extends EventProcessor> Optional<T>	getProcessor(String name) Returns the Event Processor with the given name, if present.
List<EventProcessor>	getProcessors() Returns a list of Event Processors that have been initialized.
void	initialize(Configuration config) Initialize the module configuration using the given global config
List<MessageHandlerInterceptor<? super EventMessage<?>>>	interceptorsFor(Configuration configuration, String processorName) Returns the list of Message Handler Interceptors registered for the given processorName.
EventHandlingConfiguration	registerEventHandler(Function<Configuration,Object> eventHandlerBuilder) Register an Event Handler Bean with this configuration.
EventHandlingConfiguration	registerEventProcessor(String name, EventHandlingConfiguration.EventProcessorBuilder eventProcessorBuilder) Defines the Event Processor builder for an Event Processor with the given name.
EventHandlingConfiguration	registerEventProcessorFactory(EventHandlingConfiguration.EventProcessorBuilder eventProcessorBuilder) Allows for more fine-grained definition of the Event Processor to use for each group of Event Listeners.
EventHandlingConfiguration	registerHandlerInterceptor(BiFunction<Configuration,String,MessageHandlerInterceptor<? super EventMessage<?>>> interceptorBuilder) Register the given interceptorBuilder to build an Message Handling Interceptor for Event Processors created in this configuration.
EventHandlingConfiguration	registerHandlerInterceptor(String processorName, Function<Configuration,MessageHandlerInterceptor<? super EventMessage<?>>> interceptorBuilder) Register the given interceptorBuilder to build an Message Handling Interceptor for the Event Processor with given processorName.
EventHandlingConfiguration	registerSubscribingEventProcessor(String name) Register a subscribing processor with given name that subscribes to the Event Bus.
EventHandlingConfiguration	registerSubscribingEventProcessor(String name, Function<Configuration,SubscribableMessageSource<? extends EventMessage<?>>> messageSource) Register a subscribing processor with given name that subscribes to the given messageSource.
EventHandlingConfiguration	registerTokenStore(String name, Function<Configuration,TokenStore> tokenStore) Register the TokenStore to use for a processor of given name.
EventHandlingConfiguration	registerTrackingProcessor(String name) Register a TrackingProcessor using default configuration for the given name.
EventHandlingConfiguration	registerTrackingProcessor(String name, Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> source) Registers a TrackingProcessor using the given source to read messages from.
EventHandlingConfiguration	registerTrackingProcessor(String name, Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> source, Function<Configuration,TrackingEventProcessorConfiguration> processorConfiguration, Function<Configuration,SequencingPolicy<? super EventMessage<?>>> sequencingPolicy) Registers a TrackingProcessor with the given name, reading from the given source and using the given processorConfiguration.
EventHandlingConfiguration	registerTrackingProcessor(String name, Function<Configuration,TrackingEventProcessorConfiguration> processorConfiguration, Function<Configuration,SequencingPolicy<? super EventMessage<?>>> sequencingPolicy) Registers a TrackingProcessor with the given name, reading from the Event Bus (or Store) from the main configuration and using the given processorConfiguration.
void	shutdown() Invoked prior to shutdown of the application.
void	start() Invoked when the Configuration is started.
EventHandlingConfiguration	usingTrackingProcessors() Configure the use of Tracking Event Processors, instead of the default Subscribing ones.
EventHandlingConfiguration	usingTrackingProcessors(Function<Configuration,TrackingEventProcessorConfiguration> config, Function<Configuration,SequencingPolicy<? super EventMessage<?>>> sequencingPolicy) Configure the use of Tracking Event Processors, instead of the default Subscribing ones.

Methods inherited from class java.lang.Object

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

Constructor Detail

EventHandlingConfiguration

public EventHandlingConfiguration()

Creates a default configuration for an Event Handling module that creates a SubscribingEventProcessor instance for all Event Handlers that have the same Processing Group name. The Processing Group name is determined by inspecting the ProcessingGroup annotation; if no annotation is present, the package name is used as the Processing Group name. This default behavior can be overridden in the instance returned.

At a minimum, the Event Handler beans need to be registered before this component is useful.

Method Detail

interceptorsFor

public List<MessageHandlerInterceptor<? super EventMessage<?>>> interceptorsFor(Configuration configuration,
                                                                                String processorName)

Returns the list of Message Handler Interceptors registered for the given processorName.

Parameters:

configuration - The main configuration

processorName - The name of the processor to retrieve interceptors for

Returns:

a list of Interceptors

See Also:

registerHandlerInterceptor(BiFunction), registerHandlerInterceptor(String, Function)

usingTrackingProcessors

public EventHandlingConfiguration usingTrackingProcessors()

Configure the use of Tracking Event Processors, instead of the default Subscribing ones. Tracking processors work in their own thread(s), making processing asynchronous from the publication process.

The processor will use the TokenStore implementation provided in the global Configuration, and will default to an InMemoryTokenStore when no Token Store was defined. Note that it is not recommended to use the in-memory TokenStore in a production environment.

The processors will use the a TrackingEventProcessorConfiguration registered with the configuration, or otherwise to a single threaded configuration (which means the processor will run in a single Thread and a batch size of 1).

Returns:

this EventHandlingConfiguration instance for further configuration

usingTrackingProcessors

public EventHandlingConfiguration usingTrackingProcessors(Function<Configuration,TrackingEventProcessorConfiguration> config,
                                                          Function<Configuration,SequencingPolicy<? super EventMessage<?>>> sequencingPolicy)

Configure the use of Tracking Event Processors, instead of the default Subscribing ones. Tracking processors work in their own thread(s), making processing asynchronous from the publication process.

The processor will use the TokenStore implementation provided in the global Configuration, and will default to an InMemoryTokenStore when no Token Store was defined. Note that it is not recommended to use the in-memory TokenStore in a production environment.

Parameters:

config - The configuration for the processors to use

sequencingPolicy - The policy for processing events sequentially

Returns:

this EventHandlingConfiguration instance for further configuration

registerTrackingProcessor

public EventHandlingConfiguration registerTrackingProcessor(String name)

Register a TrackingProcessor using default configuration for the given name. Unlike usingTrackingProcessors(), this method will not default all processors to tracking, but instead only use tracking for event handler that have been assigned to the processor with given name.

Events will be read from the EventBus (or EventStore) registered with the main configuration

Parameters:

name - The name of the processor

Returns:

this EventHandlingConfiguration instance for further configuration

registerTrackingProcessor

public EventHandlingConfiguration registerTrackingProcessor(String name,
                                                            Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> source)

Registers a TrackingProcessor using the given source to read messages from.

Parameters:

name - The name of the TrackingProcessor

source - The source of messages for this processor

Returns:

this EventHandlingConfiguration instance for further configuration

registerTrackingProcessor

public EventHandlingConfiguration registerTrackingProcessor(String name,
                                                            Function<Configuration,TrackingEventProcessorConfiguration> processorConfiguration,
                                                            Function<Configuration,SequencingPolicy<? super EventMessage<?>>> sequencingPolicy)

Registers a TrackingProcessor with the given name, reading from the Event Bus (or Store) from the main configuration and using the given processorConfiguration. The given sequencingPolicy defines the policy for events that need to be executed sequentially.

Parameters:

name - The name of the Tracking Processor

processorConfiguration - The configuration for the processor

sequencingPolicy - The sequencing policy to apply when processing events in parallel

Returns:

this EventHandlingConfiguration instance for further configuration

registerTrackingProcessor

public EventHandlingConfiguration registerTrackingProcessor(String name,
                                                            Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> source,
                                                            Function<Configuration,TrackingEventProcessorConfiguration> processorConfiguration,
                                                            Function<Configuration,SequencingPolicy<? super EventMessage<?>>> sequencingPolicy)

Registers a TrackingProcessor with the given name, reading from the given source and using the given processorConfiguration. The given sequencingPolicy defines the policy for events that need to be executed sequentially.

Parameters:

name - The name of the Tracking Processor

source - The source to read Events from

processorConfiguration - The configuration for the processor

sequencingPolicy - The sequencing policy to apply when processing events in parallel

Returns:

this EventHandlingConfiguration instance for further configuration

registerEventProcessorFactory

public EventHandlingConfiguration registerEventProcessorFactory(EventHandlingConfiguration.EventProcessorBuilder eventProcessorBuilder)

Allows for more fine-grained definition of the Event Processor to use for each group of Event Listeners. The given builder is expected to create a fully initialized Event Processor implementation based on the name and list of event handler beans. The builder also received the global configuration instance, from which it can retrieve components.

Note that the processor must be initialized, but shouldn't be started yet. The processor's EventProcessor.start() method is invoked when the global configuration is started.

Parameters:

eventProcessorBuilder - The builder function for the Event Processor

Returns:

this EventHandlingConfiguration instance for further configuration

registerEventProcessor

public EventHandlingConfiguration registerEventProcessor(String name,
                                                         EventHandlingConfiguration.EventProcessorBuilder eventProcessorBuilder)

Defines the Event Processor builder for an Event Processor with the given name. Event Processors registered using this method have priority over those defined in registerEventProcessorFactory(EventProcessorBuilder).

The given builder is expected to create a fully initialized Event Processor implementation based on the name and list of event handler beans. The builder also received the global configuration instance, from which it can retrieve components.

Note that the processor must be initialized, but shouldn't be started yet. The processor's EventProcessor.start() method is invoked when the global configuration is started.

Parameters:

name - The name of the Event Processor for which to use this builder

eventProcessorBuilder - The builder function for the Event Processor

Returns:

this EventHandlingConfiguration instance for further configuration

registerHandlerInterceptor

public EventHandlingConfiguration registerHandlerInterceptor(String processorName,
                                                             Function<Configuration,MessageHandlerInterceptor<? super EventMessage<?>>> interceptorBuilder)

Register the given interceptorBuilder to build an Message Handling Interceptor for the Event Processor with given processorName.

The interceptorBuilder may return null, in which case the return value is ignored.

Note that a CorrelationDataInterceptor is registered by default. To change correlation data attached to messages, see Configurer.configureCorrelationDataProviders(Function).

Parameters:

processorName - The name of the processor to register the interceptor on

interceptorBuilder - The function providing the interceptor to register, or null

Returns:

this EventHandlingConfiguration instance for further configuration

registerHandlerInterceptor

public EventHandlingConfiguration registerHandlerInterceptor(BiFunction<Configuration,String,MessageHandlerInterceptor<? super EventMessage<?>>> interceptorBuilder)

Register the given interceptorBuilder to build an Message Handling Interceptor for Event Processors created in this configuration.

The interceptorBuilder is invoked once for each processor created, and may return null, in which case the return value is ignored.

Note that a CorrelationDataInterceptor is registered by default. To change correlation data attached to messages, see Configurer.configureCorrelationDataProviders(Function).

Parameters:

interceptorBuilder - The builder function that provides an interceptor for each available processor

Returns:

this EventHandlingConfiguration instance for further configuration

byDefaultAssignTo

public EventHandlingConfiguration byDefaultAssignTo(String name)

Registers the Event Processor name to assign Event Handler beans to when no other, more explicit, rule matches and no ProcessingGroup annotation is found.

Parameters:

name - The Event Processor name to assign Event Handlers to

Returns:

this EventHandlingConfiguration instance for further configuration

byDefaultAssignTo

public EventHandlingConfiguration byDefaultAssignTo(Function<Object,String> assignmentFunction)

Registers a function that defines the Event Processor name to assign Event Handler beans to when no other, more explicit, rule matches and no ProcessingGroup annotation is found.

Parameters:

assignmentFunction - The function that returns a Processor Name for each Event Handler bean

Returns:

this EventHandlingConfiguration instance for further configuration

assignHandlersMatching

public EventHandlingConfiguration assignHandlersMatching(String name,
                                                         Predicate<Object> criteria)

Configures a rule to assign Event Handler beans that match the given criteria to the Event Processor with given name, with neutral priority (value 0).

Note that, when beans match multiple criteria for different processors with equal priority, the outcome is undefined.

Parameters:

name - The name of the Event Processor to assign matching Event Handlers to

criteria - The criteria for Event Handler to match

Returns:

this EventHandlingConfiguration instance for further configuration

assignHandlersMatching

public EventHandlingConfiguration assignHandlersMatching(String name,
                                                         int priority,
                                                         Predicate<Object> criteria)

Configures a rule to assign Event Handler beans that match the given criteria to the Event Processor with given name, with given priority. Rules with higher value of priority take precedence over those with a lower value.

Note that, when beans match multiple criteria for different processors with equal priority, the outcome is undefined.

Parameters:

name - The name of the Event Processor to assign matching Event Handlers to

priority - The priority for this rule

criteria - The criteria for Event Handler to match

Returns:

this EventHandlingConfiguration instance for further configuration

registerEventHandler

public EventHandlingConfiguration registerEventHandler(Function<Configuration,Object> eventHandlerBuilder)

Register an Event Handler Bean with this configuration. The builder function receives the global Configuration and is expected to return a fully initialized Event Handler bean, which is to be assigned to an Event Processor using configured rules.

Parameters:

eventHandlerBuilder - The builder function for the Event Handler bean

Returns:

this EventHandlingConfiguration instance for further configuration

initialize

public void initialize(Configuration config)

Description copied from interface: ModuleConfiguration

Initialize the module configuration using the given global config

Specified by:

initialize in interface ModuleConfiguration

Parameters:

config - the global configuration, providing access to generic components

start

public void start()

Description copied from interface: ModuleConfiguration

Invoked when the Configuration is started.

Specified by:

start in interface ModuleConfiguration

See Also:

Configuration.start()

shutdown

public void shutdown()

Description copied from interface: ModuleConfiguration

Invoked prior to shutdown of the application.

Specified by:

shutdown in interface ModuleConfiguration

See Also:

Configuration.shutdown()

registerSubscribingEventProcessor

public EventHandlingConfiguration registerSubscribingEventProcessor(String name)

Register a subscribing processor with given name that subscribes to the Event Bus.

Parameters:

name - The name of the Event Processor

Returns:

this EventHandlingConfiguration instance for further configuration

registerSubscribingEventProcessor

public EventHandlingConfiguration registerSubscribingEventProcessor(String name,
                                                                    Function<Configuration,SubscribableMessageSource<? extends EventMessage<?>>> messageSource)

Register a subscribing processor with given name that subscribes to the given messageSource. This allows the use of standard Subscribing Processors that listen to another source than the Event Bus.

Parameters:

name - The name of the Event Processor

messageSource - The source the processor should read from

Returns:

this EventHandlingConfiguration instance for further configuration

registerTokenStore

public EventHandlingConfiguration registerTokenStore(String name,
                                                     Function<Configuration,TokenStore> tokenStore)

Register the TokenStore to use for a processor of given name.

If no explicit TokenStore implementation is available for a Processor, it is taken from the main Configuration.

Note that this configuration is ignored if the processor with given name isn't a Tracking Processor.

Parameters:

name - The name of the processor to configure the token store for

tokenStore - The function providing the TokenStore based on a given Configuration

Returns:

this EventHandlingConfiguration instance for further configuration

getProcessors

public List<EventProcessor> getProcessors()

Returns a list of Event Processors that have been initialized. Note that an empty list may be returned if this configuration hasn't been initialized yet.

Returns:

a read-only list of processors initialized in this configuration.

getProcessor

public <T extends EventProcessor> Optional<T> getProcessor(String name)

Returns the Event Processor with the given name, if present. This method also returns an unresolved optional if the Processor was configured, but it hasn't been assigned any Event Handlers.

Parameters:

name - The name of the processor to return

Returns:

an Optional referencing the processor, if present.

configureMessageMonitor

public EventHandlingConfiguration configureMessageMonitor(String name,
                                                          Function<Configuration,MessageMonitor<Message<?>>> messageMonitorBuilder)

Configures the builder function to create the Message Monitor for the EventProcessor of the given name. This overrides any Message Monitor configured through Configurer.

Parameters:

name - The name of the event processor

messageMonitorBuilder - The builder function to use

Returns:

this EventHandlingConfiguration instance for further configuration

configureMessageMonitor

public EventHandlingConfiguration configureMessageMonitor(String name,
                                                          MessageMonitorFactory messageMonitorFactory)

Configures the factory to create the Message Monitor for the EventProcessor of the given name. This overrides any Message Monitor configured through Configurer.

Parameters:

name - The name of the event processor

messageMonitorFactory - The factory to use

Returns:

this EventHandlingConfiguration instance for further configuration