Subscribing Event Processors
The
SubscribingEventProcessor
, or Subscribing Processor for short, is a type of Event Processor. As any Event Processor, it serves as the technical aspect to handle events by invoking the event handlers written in an Axon application.The Subscribing Processor defines itself by receiving the events from a
SubscribableMessageSource
. The SubscribableMessageSource
is an infrastructure component to register a Subscribing Processor too.After registration to the
SubscribableMessageSource
, the message source gives the events to the SubscribingEventProcessor
in the order they are received. Examples of a SubscribableMessageSource
are the EventBus
or the AMQP Extension. Both the EventBus
and AMQP Extension are simple message bus solutions for events.The simple bus solution makes the
SubscribableMessageSource
and thus the Subscribing Processor an approach to only receive current events. Operations like replaying are, therefore, not an option for any Subscribing Processor as long as the SubscribableMessageSource
follows this paradigm.Furthermore, the message source will use the same thread that receives the events to invoke the registered Subscribing Processors. When the
EventBus
is, for example, used as the message source, this means that the event publishing thread is the same one handling the event in the Subscribing Processor.Although this approach deserves a spot within the framework, most scenarios require further decoupling of components by separating the threads as well. When, for example, an application requires event processing parallelization to get a higher performance, this can be a blocker. This predicament is why the
SubscribingEventProcessor
is not the default in Axon Framework.Instead, the "Tracking Event Processor" (a Streaming Processor implementation) takes up that role. It provides greater flexibility for developers for configuring the event processor in greater detail.
Subscribing Processor Use CasesAlthough theSubscribingEventProcessor
does not support easy parallelization or replays, there are still scenarios when it is beneficial.
Other than configuring that an app uses a Subscribing Event Processor, everything is covered here. Firstly, to specify that a new Event Processors should default to a
SubscribingEventProcessor
, you can use the usingSubscribingEventProcessors
method:Axon Configuration API
Spring Boot AutoConfiguration
public class AxonConfig {
// omitting other configuration methods...
public void configureProcessorDefault(EventProcessingConfigurer processingConfigurer) {
processingConfigurer.usingSubscribingEventProcessors();
}
}
@Configuration
public class AxonConfig {
// omitting other configuration methods...
@Bean
public ConfigurerModule processorDefaultConfigurerModule() {
return configurer -> configurer.eventProcessing(EventProcessingConfigurer::usingSubscribingEventProcessors);
}
}
For a specific Event Processor to be a Subscribing instance,
registerSubscribingEventProcessor
is used:Axon Configuration API
Spring Boot AutoConfiguration - Java
Spring Boot AutoConfiguration - Properties file
public class AxonConfig {
// omitting other configuration methods...
public void configureSubscribingProcessors(EventProcessingConfigurer processingConfigurer) {
// To configure a processor to be subscribing ...
processingConfigurer.registerSubscribingEventProcessor("my-processor")
// ... to define a specific SubscribableMessageSource ...
.registerSubscribingEventProcessor("my-processor", conf -> /* create/return SubscribableMessageSource */);
}
}
@Configuration
public class AxonConfig {
// omitting other configuration methods...
@Bean
public ConfigurerModule subscribingProcessorsConfigurerModule() {
return configurer -> configurer.eventProcessing(
// To configure a processor to be subscribing ...
processingConfigurer -> processingConfigurer.registerSubscribingEventProcessor("my-processor")
// ... to define a specific SubscribableMessageSource ...
.registerSubscribingEventProcessor(
"my-processor",
conf -> /* create/return SubscribableMessageSource */
)
);
}
}
A properties file allows the configuration of some fields on an Event Processor. Do note that the Java configuration provides more degrees of freedom.
axon.eventhandling.processors.my-processor.mode=subscribing
axon.eventhandling.processors.my-processor.source=eventBus
If the name of an event processor contains periods
.
, use the map notation:axon.eventhandling.processors[my.processor].mode=subscribing
axon.eventhandling.processors[my.processor].source=eventBus
Whenever the error handler rethrows an exception, the
SubscribingEventProcessor
will have it bubble up to the publishing component of the event. Providing the exception to the event publisher allows the publishing component to deal with it accordingly.Last modified 3mo ago