org.axonframework.eventhandling

Class TrackingEventProcessor

java.lang.Object

org.axonframework.eventhandling.AbstractEventProcessor

org.axonframework.eventhandling.TrackingEventProcessor

All Implemented Interfaces:

EventProcessor

public class TrackingEventProcessor
extends AbstractEventProcessor

EventProcessor implementation that tracks events from a StreamableMessageSource.

A supplied TokenStore allows the EventProcessor to keep track of its position in the event log. After processing an event batch the EventProcessor updates its tracking token in the TokenStore.

A TrackingEventProcessor is able to continue processing from the last stored token when it is restarted. It is also capable of replaying events from any starting token. To replay the entire event log simply remove the tracking token of this processor from the TokenStore. To replay from a given point first update the entry for this processor in the TokenStore before starting this processor.

Note, the name of the EventProcessor is used to obtain the tracking token from the TokenStore, so take care when renaming a TrackingEventProcessor.

Author:

Rene de Waele, Christophe Bouhier

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	TrackingEventProcessor.State

Constructor Summary

Constructors

Constructor and Description
TrackingEventProcessor(String name, EventHandlerInvoker eventHandlerInvoker, StreamableMessageSource<TrackedEventMessage<?>> messageSource, TokenStore tokenStore, TransactionManager transactionManager) Initializes an EventProcessor with given name that subscribes to the given messageSource for events.
TrackingEventProcessor(String name, EventHandlerInvoker eventHandlerInvoker, StreamableMessageSource<TrackedEventMessage<?>> messageSource, TokenStore tokenStore, TransactionManager transactionManager, MessageMonitor<? super EventMessage<?>> messageMonitor) Initializes an EventProcessor with given name that subscribes to the given messageSource for events.
TrackingEventProcessor(String name, EventHandlerInvoker eventHandlerInvoker, StreamableMessageSource<TrackedEventMessage<?>> messageSource, TokenStore tokenStore, TransactionManager transactionManager, MessageMonitor<? super EventMessage<?>> messageMonitor, RollbackConfiguration rollbackConfiguration, ErrorHandler errorHandler, TrackingEventProcessorConfiguration config) Initializes an EventProcessor with given name that subscribes to the given messageSource for events.

Method Summary

Modifier and Type	Method and Description
int	activeProcessorThreads() Returns an approximation of the number of threads currently processing events.
protected TrackingEventProcessor.State	getState() Get the state of the event processor.
boolean	isError() Indicates whether the processor has been paused due to an error.
boolean	isRunning() Indicates whether this processor is currently running (i.e.
void	pause() Stops processing if it currently running, but doesn't stop free up the processing thread.
protected void	processingLoop(Segment segment) Fetch and process event batches continuously for as long as the processor is not shutting down.
void	shutDown() Shut down the processor.
void	start() Start this processor.
protected void	startSegmentWorkers() Starts the workers for a number of segments.

Methods inherited from class org.axonframework.eventhandling.AbstractEventProcessor

canHandle, getName, processInUnitOfWork, registerInterceptor, reportIgnored, toString

Methods inherited from class java.lang.Object

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

Constructor Detail

TrackingEventProcessor

public TrackingEventProcessor(String name,
                              EventHandlerInvoker eventHandlerInvoker,
                              StreamableMessageSource<TrackedEventMessage<?>> messageSource,
                              TokenStore tokenStore,
                              TransactionManager transactionManager)

Initializes an EventProcessor with given name that subscribes to the given messageSource for events. Actual handling of event messages is deferred to the given eventHandlerInvoker.

The EventProcessor is initialized with a batch size of 1, a PropagatingErrorHandler, a RollbackConfigurationType.ANY_THROWABLE and a NoOpMessageMonitor.

Parameters:

name - The name of the event processor

eventHandlerInvoker - The component that handles the individual events

messageSource - The message source (e.g. Event Bus) which this event processor will track

tokenStore - Used to store and fetch event tokens that enable the processor to track its progress

transactionManager - The transaction manager used when processing messages

TrackingEventProcessor

public TrackingEventProcessor(String name,
                              EventHandlerInvoker eventHandlerInvoker,
                              StreamableMessageSource<TrackedEventMessage<?>> messageSource,
                              TokenStore tokenStore,
                              TransactionManager transactionManager,
                              MessageMonitor<? super EventMessage<?>> messageMonitor)

Initializes an EventProcessor with given name that subscribes to the given messageSource for events. Actual handling of event messages is deferred to the given eventHandlerInvoker.

The EventProcessor is initialized with a batch size of 1, a PropagatingErrorHandler and a RollbackConfigurationType.ANY_THROWABLE.

Parameters:

name - The name of the event processor

eventHandlerInvoker - The component that handles the individual events

messageSource - The message source (e.g. Event Bus) which this event processor will track

tokenStore - Used to store and fetch event tokens that enable the processor to track its progress

transactionManager - The transaction manager used when processing messages

messageMonitor - Monitor to be invoked before and after event processing

TrackingEventProcessor

public TrackingEventProcessor(String name,
                              EventHandlerInvoker eventHandlerInvoker,
                              StreamableMessageSource<TrackedEventMessage<?>> messageSource,
                              TokenStore tokenStore,
                              TransactionManager transactionManager,
                              MessageMonitor<? super EventMessage<?>> messageMonitor,
                              RollbackConfiguration rollbackConfiguration,
                              ErrorHandler errorHandler,
                              TrackingEventProcessorConfiguration config)

Initializes an EventProcessor with given name that subscribes to the given messageSource for events. Actual handling of event messages is deferred to the given eventHandlerInvoker.

Parameters:

name - The name of the event processor

eventHandlerInvoker - The component that handles the individual events

messageSource - The message source (e.g. Event Bus) which this event processor will track

tokenStore - Used to store and fetch event tokens that enable the processor to track its progress

transactionManager - The transaction manager used when processing messages

messageMonitor - Monitor to be invoked before and after event processing

rollbackConfiguration - Determines rollback behavior of the UnitOfWork while processing a batch of events

errorHandler - Invoked when a UnitOfWork is rolled back during processing

config - The configuration for the event processor.

Method Detail

start

public void start()

Start this processor. The processor will open an event stream on its message source in a new thread using StreamableMessageSource.openStream(TrackingToken). The TrackingToken used to open the stream will be fetched from the TokenStore.

processingLoop

protected void processingLoop(Segment segment)

Fetch and process event batches continuously for as long as the processor is not shutting down. The processor will process events in batches. The maximum size of size of each event batch is configurable.

Events with the same tracking token (which is possible as result of upcasting) should always be processed in the same batch. In those cases the batch size may be larger than the one configured.

Parameters:

segment - The Segment of the Stream that should be processed.

pause

public void pause()

Stops processing if it currently running, but doesn't stop free up the processing thread. If the processor is not running, the state isn't changed.

isRunning

public boolean isRunning()

Indicates whether this processor is currently running (i.e. consuming events from a stream).

Returns:

true when running, otherwise false

isError

public boolean isError()

Indicates whether the processor has been paused due to an error. In such case, the processor has forcefully paused, as it wasn't able to automatically recover.

Note that this method also returns false when the processor was paused using pause().

Returns:

true when paused due to an error, otherwise false

shutDown

public void shutDown()

Shut down the processor.

activeProcessorThreads

public int activeProcessorThreads()

Returns an approximation of the number of threads currently processing events.

Returns:

an approximation of the number of threads currently processing events

getState

protected TrackingEventProcessor.State getState()

Get the state of the event processor. This will indicate whether or not the processor has started or is shutting down.

Returns:

the processor state

startSegmentWorkers

protected void startSegmentWorkers()

Starts the workers for a number of segments. When only the root segment exists in the TokenStore, it will be split in multiple segments as configured by the TrackingEventProcessorConfiguration.andInitialSegmentsCount(int), otherwise the existing segments in the TokenStore will be used.

An attempt will be made to instantiate a TrackingSegmentWorker for each segment. This will succeed when the number of threads matches the requested segments. The number of active threads can be configured with TrackingEventProcessorConfiguration.forParallelProcessing(int). When insufficient threads are available to serve the number of segments, it will result in some segments not being processed.