public class EventProcessingModule extends Object implements ModuleConfiguration, EventProcessingConfiguration, EventProcessingConfigurer
EventProcessingConfiguration and takes care of module lifecycle.EventProcessingConfigurer.EventProcessorBuilder, EventProcessingConfigurer.PooledStreamingProcessorConfiguration| Constructor and Description |
|---|
EventProcessingModule() |
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, waitisType, phase, shutdown, start, unwrapeventProcessor, eventProcessor, eventProcessorByProcessingGroup, sagaConfiguration, sagaEventProcessorassignHandlerInstancesMatching, assignHandlerTypesMatching, byDefaultAssignTo, registerMessageMonitor, registerPooledStreamingEventProcessor, registerPooledStreamingEventProcessor, registerSaga, registerSubscribingEventProcessor, registerTrackingEventProcessorpublic void initialize(Configuration configuration)
ModuleConfigurationconfig. Any specific start up or shut down
processes should be added here by using the provided config and invoke Configuration.onStart(int,
LifecycleHandler) and Configuration.onShutdown(int, LifecycleHandler) respectively.initialize in interface ModuleConfigurationconfiguration - the global configuration, providing access to generic componentspublic <T extends EventProcessor> Optional<T> eventProcessorByProcessingGroup(String processingGroup)
EventProcessingConfigurationEventProcessor by it's processingGroup.eventProcessorByProcessingGroup in interface EventProcessingConfigurationT - the type of the expected EventProcessorprocessingGroup - a String specifying the processing group of an EventProcessorOptional referencing the EventProcessorpublic Map<String,EventProcessor> eventProcessors()
EventProcessingConfigurationEventProcessors.eventProcessors in interface EventProcessingConfigurationMap of registered EventProcessors within this configuration with the processor names as
keyspublic String sagaProcessingGroup(Class<?> sagaType)
EventProcessingConfigurationsagaType.sagaProcessingGroup in interface EventProcessingConfigurationsagaType - the type of Sagapublic List<MessageHandlerInterceptor<? super EventMessage<?>>> interceptorsFor(String processorName)
EventProcessingConfigurationinterceptorsFor in interface EventProcessingConfigurationprocessorName - a String specifying a processing groupList of MessageHandlerInterceptors for a processor with given processorNamepublic ListenerInvocationErrorHandler listenerInvocationErrorHandler(String processingGroup)
EventProcessingConfigurationListenerInvocationErrorHandler tied to the given processingGroup.listenerInvocationErrorHandler in interface EventProcessingConfigurationprocessingGroup - a String specifying a processing groupListenerInvocationErrorHandler belonging to the given processingGrouppublic SequencingPolicy<? super EventMessage<?>> sequencingPolicy(String processingGroup)
EventProcessingConfigurationSequencingPolicy tied to the given processingGroup.sequencingPolicy in interface EventProcessingConfigurationprocessingGroup - a String specifying a processing groupSequencingPolicy belonging to the given processingGrouppublic RollbackConfiguration rollbackConfiguration(String processorName)
EventProcessingConfigurationRollbackConfiguration tied to the given processorName.rollbackConfiguration in interface EventProcessingConfigurationprocessorName - a String specifying a processing groupRollbackConfiguration belonging to the given processorNamepublic ErrorHandler errorHandler(String processorName)
EventProcessingConfigurationErrorHandler tied to the given processorName.errorHandler in interface EventProcessingConfigurationprocessorName - a String specifying a processing groupErrorHandler belonging to the given processorNamepublic SagaStore sagaStore()
EventProcessingConfigurationSagaStore registered within this configuration.sagaStore in interface EventProcessingConfigurationSagaStore registered within this configurationpublic List<SagaConfiguration<?>> sagaConfigurations()
EventProcessingConfigurationList of SagaConfigurations registered within this configuration.sagaConfigurations in interface EventProcessingConfigurationList of SagaConfigurations registered within this configurationpublic MessageMonitor<? super Message<?>> messageMonitor(Class<?> componentType, String eventProcessorName)
EventProcessingConfigurationMessageMonitor set to the given componentType and componentName registered
within this configuration.messageMonitor in interface EventProcessingConfigurationcomponentType - a Class type of component to be monitoredeventProcessorName - a String specifying the name of the component to be monitoredMessageMonitor registered to the given componentType and componentNamepublic TokenStore tokenStore(String processorName)
EventProcessingConfigurationTokenStore tied to the given processorName.tokenStore in interface EventProcessingConfigurationprocessorName - a String specifying a event processorTokenStore belonging to the given processorNamepublic TransactionManager transactionManager(String processorName)
EventProcessingConfigurationTransactionManager tied to the given processorName.transactionManager in interface EventProcessingConfigurationprocessorName - a String specifying a processing groupTransactionManagerbelonging to the given processorNamepublic <T> EventProcessingConfigurer registerSaga(Class<T> sagaType, Consumer<SagaConfigurer<T>> sagaConfigurer)
EventProcessingConfigurerregisterSaga in interface EventProcessingConfigurerT - The type of Saga to configuresagaType - The type of Saga to configuresagaConfigurer - a function providing modifications on top of the defaul configurationEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerSagaStore(Function<Configuration,SagaStore> sagaStoreBuilder)
EventProcessingConfigurerregisterSagaStore in interface EventProcessingConfigurersagaStoreBuilder - a Function that builds a SagaStoreEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerEventHandler(Function<Configuration,Object> eventHandlerBuilder)
EventProcessingConfigurerFunction that builds an Event Handler instance.registerEventHandler in interface EventProcessingConfigurereventHandlerBuilder - a Function that builds an Event Handler instanceEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerDefaultListenerInvocationErrorHandler(Function<Configuration,ListenerInvocationErrorHandler> listenerInvocationErrorHandlerBuilder)
EventProcessingConfigurerFunction that builds the default ListenerInvocationErrorHandler.
Defaults to a LoggingErrorHandler.registerDefaultListenerInvocationErrorHandler in interface EventProcessingConfigurerlistenerInvocationErrorHandlerBuilder - a Function that builds the default
ListenerInvocationErrorHandlerEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerListenerInvocationErrorHandler(String processingGroup, Function<Configuration,ListenerInvocationErrorHandler> listenerInvocationErrorHandlerBuilder)
EventProcessingConfigurerregisterListenerInvocationErrorHandler in interface EventProcessingConfigurerprocessingGroup - a String specifying the name of a processing grouplistenerInvocationErrorHandlerBuilder - a Function that builds ListenerInvocationErrorHandlerEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer configureDefaultStreamableMessageSource(Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> defaultSource)
EventProcessingConfigurerStreamableMessageSource to use for Tracking Event Processors if none was explicitly
provided. Defaults to the Event Bus (or Store) available in the Configuration.
Note that the configuration of a default source does not change how the decision is made to select the
type of processor. Unless explicitly specified using EventProcessingConfigurer.usingSubscribingEventProcessors() or
EventProcessingConfigurer.usingTrackingEventProcessors(), the default is dependent on the type of Message Source the Event Bus
provides. If the Event Bus supports Tracking Processors, that is the default, otherwise Subscribing Event
Processors are the default.
configureDefaultStreamableMessageSource in interface EventProcessingConfigurerdefaultSource - a Function that defines the Message source to useEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer configureDefaultSubscribableMessageSource(Function<Configuration,SubscribableMessageSource<EventMessage<?>>> defaultSource)
EventProcessingConfigurerSubscribableMessageSource to use for Subscribing Event Processors if none was explicitly
provided. Defaults to the Event Bus (or Store) available in the Configuration.
Note that the configuration of a default source does not change how the decision is made to select the
type of processor. Unless explicitly specified using EventProcessingConfigurer.usingSubscribingEventProcessors() or
EventProcessingConfigurer.usingTrackingEventProcessors(), the default is dependent on the type of Message Source the Event Bus
provides. If the Event Bus supports Tracking Processors, that is the default, otherwise Subscribing Event
Processors are the default.
configureDefaultSubscribableMessageSource in interface EventProcessingConfigurerdefaultSource - a Function that defines the Message source to useEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerTrackingEventProcessor(String name, Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> source)
EventProcessingConfigurerregisterTrackingEventProcessor in interface EventProcessingConfigurername - a String specifying the name of the TrackingEventProcessor
being registeredsource - a Function that builds a StreamableMessageSourceEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerTrackingEventProcessor(String name, Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> source, Function<Configuration,TrackingEventProcessorConfiguration> processorConfiguration)
EventProcessingConfigurerTrackingEventProcessor with given name, source and processorConfiguration within this Configurer.registerTrackingEventProcessor in interface EventProcessingConfigurername - a String specifying the name of the TrackingEventProcessor
being registeredsource - a Function that builds StreamableMessageSourceprocessorConfiguration - a Function that builds a TrackingEventProcessorConfigurationEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerEventProcessorFactory(EventProcessingConfigurer.EventProcessorBuilder eventProcessorBuilder)
EventProcessingConfigurerEventProcessor. This is the EventProcessingConfigurer.EventProcessorBuilder to
be used when there is no specific builder for given processor name.registerEventProcessorFactory in interface EventProcessingConfigurereventProcessorBuilder - a Function that builds an EventProcessorEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerEventProcessor(String name, EventProcessingConfigurer.EventProcessorBuilder eventProcessorBuilder)
EventProcessingConfigurerEventProcessingConfigurer.EventProcessorBuilder for the given processor name.registerEventProcessor in interface EventProcessingConfigurername - a String specifying the name of the EventProcessor being registeredeventProcessorBuilder - a Function that builds an EventProcessorEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerTokenStore(String processingGroup, Function<Configuration,TokenStore> tokenStore)
EventProcessingConfigurerregisterTokenStore in interface EventProcessingConfigurerprocessingGroup - a String specifying the name of a event processortokenStore - a Function that builds a TokenStoreEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerTokenStore(Function<Configuration,TokenStore> tokenStore)
EventProcessingConfigurerFunction that builds a TokenStore to use as the default in case no explicit token
store was configured for a processor.registerTokenStore in interface EventProcessingConfigurertokenStore - a Function that builds a TokenStoreEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer usingSubscribingEventProcessors()
EventProcessingConfigurerSubscribingEventProcessor.
The default behavior depends on the EventBus available in the Configuration. If the Event Bus is a
StreamableMessageSource, processors are Tracking by default. This method must be used to force the use
of Subscribing Processors, unless specifically overridden for individual processors.
usingSubscribingEventProcessors in interface EventProcessingConfigurerEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer usingTrackingEventProcessors()
EventProcessingConfigurerTrackingEventProcessor.
The default behavior depends on the EventBus available in the Configuration. If the Event Bus is a
StreamableMessageSource, processors are Tracking by default. This method must be used to force the use
of Tracking Processors, unless specifically overridden for individual processors.
usingTrackingEventProcessors in interface EventProcessingConfigurerEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer usingPooledStreamingEventProcessors()
EventProcessingConfigurerPooledStreamingEventProcessor.
The default behavior depends on the EventBus available in the Configuration. If the EventBus is a StreamableMessageSource, processors are Tracking by default. This method must be used to
force the use of Pooled Streaming Processors, unless specifically overridden for individual processors.
usingPooledStreamingEventProcessors in interface EventProcessingConfigurerEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerSubscribingEventProcessor(String name, Function<Configuration,SubscribableMessageSource<? extends EventMessage<?>>> messageSource)
EventProcessingConfigurerregisterSubscribingEventProcessor in interface EventProcessingConfigurername - a String specyfing the name of the SubscribingEventProcessor
being registeredmessageSource - a Function that builds a SubscribableMessageSourceEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerDefaultErrorHandler(Function<Configuration,ErrorHandler> errorHandlerBuilder)
EventProcessingConfigurerregisterDefaultErrorHandler in interface EventProcessingConfigurererrorHandlerBuilder - a Function that builds an ErrorHandlerEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerErrorHandler(String eventProcessorName, Function<Configuration,ErrorHandler> errorHandlerBuilder)
EventProcessingConfigurerregisterErrorHandler in interface EventProcessingConfigurereventProcessorName - a String specifying the name of an EventProcessorerrorHandlerBuilder - a Function that builds an ErrorHandlerEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer byDefaultAssignHandlerInstancesTo(Function<Object,String> assignmentFunction)
EventProcessingConfigurerFunction that defines the Event Processing Group name to assign Event Handler beans to when
no other, more explicit, rule matches and no ProcessingGroup annotation is found.byDefaultAssignHandlerInstancesTo in interface EventProcessingConfigurerassignmentFunction - a Function that returns the Processing Group for each Event Handler beanEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer byDefaultAssignHandlerTypesTo(Function<Class<?>,String> assignmentFunction)
EventProcessingConfigurerFunction that defines the Event Processing Group name to assign Event Handler and Saga beans
to when no other, more explicit, rule matches and no ProcessingGroup annotation is found.byDefaultAssignHandlerTypesTo in interface EventProcessingConfigurerassignmentFunction - a Function that returns the Processing Group for each Event Handler or Saga
beanEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer assignHandlerInstancesMatching(String processingGroup, int priority, Predicate<Object> criteria)
EventProcessingConfigurercriteria to the Processing Group
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 processing groups with equal priority, the outcome is undefined.
assignHandlerInstancesMatching in interface EventProcessingConfigurerprocessingGroup - a String specifying the name of a processing group to assign matching Event
Handlers topriority - The priority for this rulecriteria - a Predicate defining the criteria for an Event Handler to matchEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer assignHandlerTypesMatching(String processingGroup, int priority, Predicate<Class<?>> criteria)
EventProcessingConfigurercriteria to the Processing Group
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 processing groups with equal priority, the outcome is undefined.
assignHandlerTypesMatching in interface EventProcessingConfigurerprocessingGroup - a String specifying the name of the Processing Group to assign matching Event
Handlers or Sagas topriority - an int specifying the priority of this rulecriteria - a Predicate defining the criteria for an Event Handler or Saga to matchEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer assignProcessingGroup(String processingGroup, String processorName)
EventProcessingConfigurerassignProcessingGroup in interface EventProcessingConfigurerprocessingGroup - a String specifying the processing group to be assignedprocessorName - a String specifying the processor name to assign the group toEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer assignProcessingGroup(Function<String,String> assignmentRule)
EventProcessingConfigurerassignProcessingGroup in interface EventProcessingConfigurerassignmentRule - a Function which takes a processing group and returns a processor nameEventProcessingConfigurer instance, for fluent interfacingEventProcessingConfigurer.assignProcessingGroup(String, String)public EventProcessingConfigurer registerHandlerInterceptor(String processorName, Function<Configuration,MessageHandlerInterceptor<? super EventMessage<?>>> interceptorBuilder)
EventProcessingConfigurerinterceptorBuilder to build a MessageHandlerInterceptor for the
EventProcessor with given processorName.
The interceptorBuilder may return null, in which case the return value is ignored.
registerHandlerInterceptor in interface EventProcessingConfigurerprocessorName - a String specyfing the name of the processor to register the
MessageHandlerInterceptor oninterceptorBuilder - a Function providing the MessageHandlerInterceptor to register, or
nullEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerDefaultHandlerInterceptor(BiFunction<Configuration,String,MessageHandlerInterceptor<? super EventMessage<?>>> interceptorBuilder)
EventProcessingConfigurerinterceptorBuilder as a default to build a MessageHandlerInterceptor for
EventProcessors 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.
registerDefaultHandlerInterceptor in interface EventProcessingConfigurerinterceptorBuilder - a builder Function that provides a MessageHandlerInterceptor for each
available processorEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerSequencingPolicy(String processingGroup, Function<Configuration,SequencingPolicy<? super EventMessage<?>>> policyBuilder)
EventProcessingConfigurerSequencingPolicy created by the given policyBuilder to the processing group with
given processingGroup. Any previously configured policy for the same name will be overwritten.registerSequencingPolicy in interface EventProcessingConfigurerprocessingGroup - a String specifying the name of the processing group to assign the
SequencingPolicy forpolicyBuilder - a builder Function to create the SequencingPolicy to useEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerDefaultSequencingPolicy(Function<Configuration,SequencingPolicy<? super EventMessage<?>>> policyBuilder)
EventProcessingConfigurerSequencingPolicy created by given policyBuilder to the processing groups for which
no explicit policy is defined (using EventProcessingConfigurer.registerSequencingPolicy(String, Function)).
Defaults to a SequentialPerAggregatePolicy.
registerDefaultSequencingPolicy in interface EventProcessingConfigurerpolicyBuilder - a builder Function to create the SequencingPolicy to useEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerMessageMonitorFactory(String eventProcessorName, MessageMonitorFactory messageMonitorFactory)
EventProcessingConfigurerregisterMessageMonitorFactory in interface EventProcessingConfigurereventProcessorName - a String specifying the name of an EventProcessormessageMonitorFactory - a MessageMonitorFactory used to create a MessageMonitorEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerRollbackConfiguration(String name, Function<Configuration,RollbackConfiguration> rollbackConfigurationBuilder)
EventProcessingConfigurerFunction that builds the RollbackConfiguration for given processor name.
Defaults to a RollbackConfigurationType.ANY_THROWABLEregisterRollbackConfiguration in interface EventProcessingConfigurername - a String specifying the name of an EventProcessorrollbackConfigurationBuilder - a Function that builds a RollbackConfigurationEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerTransactionManager(String name, Function<Configuration,TransactionManager> transactionManagerBuilder)
EventProcessingConfigurerregisterTransactionManager in interface EventProcessingConfigurername - a String specifying the name of an EventProcessortransactionManagerBuilder - a Function that builds a TransactionManagerEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerTrackingEventProcessorConfiguration(String name, Function<Configuration,TrackingEventProcessorConfiguration> trackingEventProcessorConfigurationBuilder)
EventProcessingConfigurerFunction that builds a TrackingEventProcessorConfiguration to be used by the EventProcessor corresponding to the given name.registerTrackingEventProcessorConfiguration in interface EventProcessingConfigurername - a String specifying the name of an EventProcessortrackingEventProcessorConfigurationBuilder - a Function that builds a TrackingEventProcessorConfigurationEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerTrackingEventProcessorConfiguration(Function<Configuration,TrackingEventProcessorConfiguration> trackingEventProcessorConfigurationBuilder)
EventProcessingConfigurerFunction that builds a TrackingEventProcessorConfiguration to use as the default.registerTrackingEventProcessorConfiguration in interface EventProcessingConfigurertrackingEventProcessorConfigurationBuilder - a Function that builds a TrackingEventProcessorConfigurationEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerPooledStreamingEventProcessor(String name, Function<Configuration,StreamableMessageSource<TrackedEventMessage<?>>> messageSource, EventProcessingConfigurer.PooledStreamingProcessorConfiguration processorConfiguration)
EventProcessingConfigurerPooledStreamingEventProcessor in this EventProcessingConfigurer. The processor will
receive the given name and use the outcome of the messageSource as the StreamableMessageSource.
The processorConfiguration will be used to further configure the PooledStreamingEventProcessor
upon construction. Note that the processorConfiguration will override any configuration set through the
EventProcessingConfigurer.registerPooledStreamingEventProcessorConfiguration(PooledStreamingProcessorConfiguration) and EventProcessingConfigurer.registerPooledStreamingEventProcessorConfiguration(String, PooledStreamingProcessorConfiguration).
registerPooledStreamingEventProcessor in interface EventProcessingConfigurername - the name of the PooledStreamingEventProcessor being registeredmessageSource - constructs a StreamableMessageSource to be used by the PooledStreamingEventProcessorprocessorConfiguration - allows further customization of the PooledStreamingEventProcessor under
construction. The given Configuration can be used to extract components and
use them in the PooledStreamingEventProcessor.BuilderEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerPooledStreamingEventProcessorConfiguration(String name, EventProcessingConfigurer.PooledStreamingProcessorConfiguration pooledStreamingProcessorConfiguration)
EventProcessingConfigurerEventProcessingConfigurer.PooledStreamingProcessorConfiguration to be used when constructing a PooledStreamingEventProcessor with name.registerPooledStreamingEventProcessorConfiguration in interface EventProcessingConfigurername - the name of an PooledStreamingEventProcessorpooledStreamingProcessorConfiguration - configuration used when constructing a PooledStreamingEventProcessor
with the given nameEventProcessingConfigurer instance, for fluent interfacingpublic EventProcessingConfigurer registerPooledStreamingEventProcessorConfiguration(EventProcessingConfigurer.PooledStreamingProcessorConfiguration pooledStreamingProcessorConfiguration)
EventProcessingConfigurerEventProcessingConfigurer.PooledStreamingProcessorConfiguration to be used when constructing every PooledStreamingEventProcessor.registerPooledStreamingEventProcessorConfiguration in interface EventProcessingConfigurerpooledStreamingProcessorConfiguration - configuration used when constructing every PooledStreamingEventProcessorEventProcessingConfigurer instance, for fluent interfacingCopyright © 2010–2022. All rights reserved.