org.axonframework.config

Class EventProcessingConfiguration

java.lang.Object

org.axonframework.config.EventProcessingConfiguration

All Implemented Interfaces:

ModuleConfiguration

public class EventProcessingConfiguration
extends Object
implements ModuleConfiguration

Non-thread safe event processing configuration. Holds a registry of EventProcessors and EventHandlerInvokers and all necessary information in order to tie those two together. After calling initialize(Configuration), all registrations (except for MessageHandlerInterceptors) are ignored.

Since:

3.3

Author:

Milan Savic

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static interface	EventProcessingConfiguration.EventProcessorBuilder Contract which defines how to build an event processor.

Constructor Summary

Constructors

Constructor and Description
EventProcessingConfiguration()

Method Summary

Modifier and Type	Method and Description
EventProcessingConfiguration	assignProcessingGroup(Function<String,String> assignmentRule) Defines a rule for assigning processing groups to processors if processing group to processor name mapping does not contain the entry.
EventProcessingConfiguration	assignProcessingGroup(String processingGroup, String processorName) Defines a mapping for assigning processing groups to processors.
EventProcessingConfiguration	configureErrorHandler(Function<Configuration,ErrorHandler> errorHandlerBuilder) Configures the default ErrorHandler for any EventProcessor.
EventProcessingConfiguration	configureErrorHandler(String name, Function<Configuration,ErrorHandler> errorHandlerBuilder) Configures a ErrorHandler for the EventProcessor of the given name.
EventProcessingConfiguration	configureMessageMonitor(String name, Function<Configuration,MessageMonitor<Message<?>>> messageMonitorBuilder) Configures the builder function to create the Message Monitor for the EventProcessor of the given name.
EventProcessingConfiguration	configureMessageMonitor(String name, MessageMonitorFactory messageMonitorFactory) Configures the factory to create the Message Monitor for the EventProcessor of the given name.
EventProcessingConfiguration	configureRollbackConfiguration(String name, Function<Configuration,RollbackConfiguration> rollbackConfigurationBuilder) Configures a RollbackConfiguration for the EventProcessor of the given name.
EventProcessingConfiguration	configureTransactionManager(String name, Function<Configuration,TransactionManager> transactionManagerBuilder) Configures a TransactionManager for the EventProcessor of the given name.
<T extends EventProcessor> Optional<T>	eventProcessor(String name) Obtains Event Processor by name.
<T extends EventProcessor> Optional<T>	eventProcessor(String name, Class<T> expectedType) Returns the Event Processor with the given name, if present and of the given expectedType.
<T extends EventProcessor> Optional<T>	eventProcessorByProcessingGroup(String processingGroup) Obtains Event Processor by processing group name.
<T extends EventProcessor> Optional<T>	eventProcessorByProcessingGroup(String processingGroup, Class<T> expectedType) Returns the Event Processor by the given processingGroup, if present and of the given expectedType.
Map<String,EventProcessor>	eventProcessors() Obtains all registered event processors.
void	initialize(Configuration config) Initialize the module configuration using the given global config
List<MessageHandlerInterceptor<? super EventMessage<?>>>	interceptorsFor(String processorName) Returns a list of interceptors for a processor with given processorName.
int	phase() Defines a phase in which this module's ModuleConfiguration.initialize(Configuration), ModuleConfiguration.start(), ModuleConfiguration.shutdown() will be invoked.
EventProcessingConfiguration	registerEventProcessor(String name, EventProcessingConfiguration.EventProcessorBuilder eventProcessorBuilder) Defines the Event Processor builder for an Event Processor with the given name.
EventProcessingConfiguration	registerEventProcessorFactory(EventProcessingConfiguration.EventProcessorBuilder eventProcessorBuilder) Allows for more fine-grained definition of the Event Processor to use for each group of Event Listeners.
EventProcessingConfiguration	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.
EventProcessingConfiguration	registerHandlerInvoker(String processingGroup, Function<Configuration,EventHandlerInvoker> eventHandlerInvokerBuilder) Registers handler invoker within this registry which will be assigned to the event processing group with given processingGroup during the initialization phase.
EventProcessingConfiguration	registerSubscribingEventProcessor(String name) Register a subscribing event processor with given name that subscribes to the Event Bus.
EventProcessingConfiguration	registerSubscribingEventProcessor(String name, Function<Configuration,SubscribableMessageSource<? extends EventMessage<?>>> messageSource) Register a subscribing event processor with given name that subscribes to the given messageSource.
EventProcessingConfiguration	registerTokenStore(String name, Function<Configuration,TokenStore> tokenStore) Register the TokenStore to use for a processor of given name.
EventProcessingConfiguration	registerTrackingEventProcessor(String name) Register a TrackingEventProcessor using default configuration for the given name.
EventProcessingConfiguration	registerTrackingEventProcessor(String name, Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> source, Function<Configuration,TrackingEventProcessorConfiguration> processorConfiguration) Registers a TrackingEventProcessor with the given name, reading from the given source and using the given processorConfiguration.
EventProcessingConfiguration	registerTrackingEventProcessor(String name, Function<Configuration,TrackingEventProcessorConfiguration> processorConfiguration) Registers a TrackingEventProcessor with the given name, reading from the Event Bus (or Store) from the main configuration and using the given processorConfiguration.
EventProcessingConfiguration	registerTrackingEventProcessorUsingSource(String name, Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> source) Registers a TrackingEventProcessor using the given source to read messages from.
void	shutdown() Invoked prior to shutdown of the application.
void	start() Invoked when the Configuration is started.
EventProcessingConfiguration	usingTrackingProcessors() Configure the use of Tracking Event Processors, instead of the default Subscribing ones.
EventProcessingConfiguration	usingTrackingProcessors(Function<Configuration,TrackingEventProcessorConfiguration> processorConfig) 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

Methods inherited from interface org.axonframework.config.ModuleConfiguration

unwrap

Constructor Detail

EventProcessingConfiguration

public EventProcessingConfiguration()

Method Detail

registerHandlerInvoker

public EventProcessingConfiguration registerHandlerInvoker(String processingGroup,
                                                           Function<Configuration,EventHandlerInvoker> eventHandlerInvokerBuilder)

Registers handler invoker within this registry which will be assigned to the event processing group with given processingGroup during the initialization phase.

Parameters:

processingGroup - The name of the Processing Group

eventHandlerInvokerBuilder - Builder which builds event handler invoker. Builder can use configuration to obtain other components

Returns:

event processing configuration for chaining purposes

assignProcessingGroup

public EventProcessingConfiguration assignProcessingGroup(String processingGroup,
                                                          String processorName)

Defines a mapping for assigning processing groups to processors.

Parameters:

processingGroup - The processing group to be assigned

processorName - The processor name to assign group to

Returns:

event processing configuration for chaining purposes

assignProcessingGroup

public EventProcessingConfiguration assignProcessingGroup(Function<String,String> assignmentRule)

Defines a rule for assigning processing groups to processors if processing group to processor name mapping does not contain the entry.

Parameters:

assignmentRule - The function which takes processing group and returns processor name

Returns:

event processing configuration for chaining purposes

See Also:

assignProcessingGroup(String, String)

registerEventProcessor

public EventProcessingConfiguration registerEventProcessor(String name,
                                                           EventProcessingConfiguration.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:

event processing configuration for chaining purposes

registerTrackingEventProcessor

public EventProcessingConfiguration registerTrackingEventProcessor(String name)

Register a TrackingEventProcessor 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:

event processing configuration for chaining purposes

registerTrackingEventProcessorUsingSource

public EventProcessingConfiguration registerTrackingEventProcessorUsingSource(String name,
                                                                              Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> source)

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

Parameters:

name - The name of the TrackingEventProcessor

source - The source of messages for this processor

Returns:

event processing configuration for chaining purposes

registerTrackingEventProcessor

public EventProcessingConfiguration registerTrackingEventProcessor(String name,
                                                                   Function<Configuration,TrackingEventProcessorConfiguration> processorConfiguration)

Registers a TrackingEventProcessor with the given name, reading from the Event Bus (or Store) from the main configuration and using the given processorConfiguration.

Parameters:

name - The name of the Tracking Processor

processorConfiguration - The configuration for the processor

Returns:

event processing configuration for chaining purposes

registerTrackingEventProcessor

public EventProcessingConfiguration registerTrackingEventProcessor(String name,
                                                                   Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> source,
                                                                   Function<Configuration,TrackingEventProcessorConfiguration> processorConfiguration)

Registers a TrackingEventProcessor with the given name, reading from the given source and using the given processorConfiguration.

Parameters:

name - The name of the Tracking Processor

source - The source to read Events from

processorConfiguration - The configuration for the processor

Returns:

event processing configuration for chaining purposes

registerSubscribingEventProcessor

public EventProcessingConfiguration registerSubscribingEventProcessor(String name)

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

Parameters:

name - The name of the Event Processor

Returns:

event processing configuration for chaining purposes

registerSubscribingEventProcessor

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

Register a subscribing event processor with given name that subscribes to the given messageSource. This allows the use of standard Subscribing Event 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:

event processing configuration for chaining purposes

usingTrackingProcessors

public EventProcessingConfiguration 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:

event processing configuration for chaining purposes

usingTrackingProcessors

public EventProcessingConfiguration usingTrackingProcessors(Function<Configuration,TrackingEventProcessorConfiguration> processorConfig)

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:

processorConfig - The configuration for the processors to use

Returns:

event processing configuration for chaining purposes

registerTokenStore

public EventProcessingConfiguration 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:

event processing configuration for chaining purposes

registerHandlerInterceptor

public EventProcessingConfiguration 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.

Parameters:

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

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

Returns:

event processing configuration for chaining purposes

interceptorsFor

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

Returns a list of interceptors for a processor with given processorName.

Parameters:

processorName - The name of the processor

Returns:

a list of interceptors for a processor with given processorName

registerEventProcessorFactory

public EventProcessingConfiguration registerEventProcessorFactory(EventProcessingConfiguration.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:

event processing configuration for chaining purposes

configureErrorHandler

public EventProcessingConfiguration configureErrorHandler(Function<Configuration,ErrorHandler> errorHandlerBuilder)

Configures the default ErrorHandler for any EventProcessor. This can be overridden per EventProcessor by calling the configureErrorHandler(String, Function) function.

Parameters:

errorHandlerBuilder - The ErrorHandler to use for the EventProcessor with the given name

Returns:

event processing configuration for chaining purposes

configureErrorHandler

public EventProcessingConfiguration configureErrorHandler(String name,
                                                          Function<Configuration,ErrorHandler> errorHandlerBuilder)

Configures a ErrorHandler for the EventProcessor of the given name. This overrides the default ErrorHandler configured through the Configurer.

Parameters:

name - The name of the event processor

errorHandlerBuilder - The ErrorHandler to use for the EventProcessor with the given name

Returns:

event processing configuration for chaining purposes

configureRollbackConfiguration

public EventProcessingConfiguration configureRollbackConfiguration(String name,
                                                                   Function<Configuration,RollbackConfiguration> rollbackConfigurationBuilder)

Configures a RollbackConfiguration for the EventProcessor of the given name. This overrides the default RollbackConfiguration configured through the Configurer.

Parameters:

name - The name of the event processor

rollbackConfigurationBuilder - The RollbackConfiguration to use for the EventProcessor with the given name

Returns:

event processing configuration for chaining purposes

configureTransactionManager

public EventProcessingConfiguration configureTransactionManager(String name,
                                                                Function<Configuration,TransactionManager> transactionManagerBuilder)

Configures a TransactionManager for the EventProcessor of the given name. This overrides the default TransactionManager configured through the Configurer.

Parameters:

name - The name of the event processor

transactionManagerBuilder - The TransactionManager to use of the EventProcessor with the given name

Returns:

event processing configuration for chaining purposes

configureMessageMonitor

public EventProcessingConfiguration 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:

event processing configuration for chaining purposes

configureMessageMonitor

public EventProcessingConfiguration 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:

event processing configuration for chaining purposes

eventProcessors

public Map<String,EventProcessor> eventProcessors()

Obtains all registered event processors. This method is to be called after Event Processor Registry is initialized.

Returns:

map of registered event processors within this registry where processor name is the key

eventProcessor

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

Obtains Event Processor by name. This method is to be called after Event Processor Registry is initialized.

Type Parameters:

T - The type of processor expected

Parameters:

name - The name of the event processor

Returns:

optional whether event processor with given name exists

eventProcessor

public <T extends EventProcessor> Optional<T> eventProcessor(String name,
                                                             Class<T> expectedType)

Returns the Event Processor with the given name, if present and of the given expectedType.

Type Parameters:

T - The type of processor expected

Parameters:

name - The name of the processor to return

expectedType - The type of processor expected

Returns:

an Optional referencing the processor, if present and of expected type

eventProcessorByProcessingGroup

public <T extends EventProcessor> Optional<T> eventProcessorByProcessingGroup(String processingGroup)

Obtains Event Processor by processing group name. This method is to be called after Event Processor Registry is initialized.

Type Parameters:

T - The type of processor expected

Parameters:

processingGroup - The name of the processing group

Returns:

an Optional referencing the processor

eventProcessorByProcessingGroup

public <T extends EventProcessor> Optional<T> eventProcessorByProcessingGroup(String processingGroup,
                                                                              Class<T> expectedType)

Returns the Event Processor by the given processingGroup, if present and of the given expectedType.

Type Parameters:

T - The type of processor expected

Parameters:

processingGroup - The name of the processing group

expectedType - The type of processor expected

Returns:

an Optional referencing the processor, if present and of expected type

phase

public int phase()

Description copied from interface: ModuleConfiguration

Defines a phase in which this module's ModuleConfiguration.initialize(Configuration), ModuleConfiguration.start(), ModuleConfiguration.shutdown() will be invoked.

Specified by:

phase in interface ModuleConfiguration

Returns:

this module's phase

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()