Reference Guide

The demands on software projects increase rapidly as time progresses. Companies no longer accept a brochure-like homepage to promote their business; they want their (web)applications to evolve together with their business. That means that not only projects and code bases become more complex, it also means that functionality is constantly added, changed and (unfortunately not enough) removed. It can be frustrating to find out that a seemingly easy-to-implement feature can require development teams to take apart an entire application. Furthermore, today's webapplications target an audience of potentially billions of people, making scalability an indisputable requirement.

Although there are many applications and frameworks around that deal with scalability issues, such as GigaSpaces and Terracotta, they share one fundamental flaw. These stacks try to solve the scalability issues while letting developers develop applications using the layered architecture they are used to. In some cases, they even prevent or severely limit the use of a real domain model, forcing all domain logic into services. Although that is faster to start building an application, eventually this approach will cause complexity to increase and development to slow down.

The Command Query Responsiblity Segregation (CQRS) pattern addresses these issues by drastically changing the way applications are architected. Instead of separating logic into separate layers, logic is separated based on whether it is changing an application's state or querying it. That means that executing commands (actions that potentially change an application's state) are executed by different components than those that query for the application's state. The most important reason for this separation is the fact that there are different technical and non-technical requirements for each of them. When commands are executed, the query components are (a)synchronously updated using events. This mechanism of updates through events, is what makes this architecture is extensible, scalable and ultimately more maintainable.

	Note
	A full explanation of CQRS is not within the scope of this document. If you would like to have more background information about CQRS, visit the Axon Framework website: www.axonframework.org. It contains links to background information.

Since CQRS is so fundamentally different than the layered-architecture which dominates the software landscape nowadays, it is quite hard to grasp. It is not uncommon for developers to walk into a few traps while trying to find their way around this architecture. That's why Axon Framework was conceived: to help developers implement CQRS applications while focussing on the business logic.

Axon Framework helps build scalable, extensible and maintainable applications by supporting developers apply the Command Query Responsibility Segregation (CQRS) architectural pattern. It does so by providing implementations of the most important building blocks, such as aggregates, repositories and event buses (the dispatching mechanism for events). Furthermore, Axon provides annotation support, which allows you to build aggregates and event listeners withouth tying your code to Axon specific logic. This allows you to focus on your business logic, instead of the plumbing, and helps you to make your code easier to test in isolation.

Axon does not, in any way, try to hide the CQRS architecture or any of its components from developers. Therefore, depending on team size, it is still advisable to have one or more developers with a thorough understanding of CQRS on each team. However, Axon does help when it comes to guaranteeing delivering events to the right event listeners and processing them concurrently and in the correct order. These multi-threading concerns are typically hard to deal with, leading to hard-to-trace bugs and sometimes complete application failure. When you have a tight deadline, you probably don't even want to care about these concerns. Axon's code is thoroughly tested to prevent these types of bugs.

The Axon Framework consists of a number of modules (jars) that provide the tools and components to build a scalable infrastructure. The Axon Core module provides the basic API's for the different components, and simple implemenatinos that provide solutions for single-JVM applications. The other modules address scalability or high-performance issues, by providing specialized building blocks.

Will each application benefit from Axon? Unfortunately not. Simple CRUD (Create, Read, Update, Delete) applications which are not expected to scale will probably not benefit from CQRS or Axon. Fortunately, there is a wide variety of applications that does benefit from Axon.

Applications that will most likely benefit from CQRS and Axon are those that show one or more of the following characteristics:

You can download the Axon Framework from our downloads page: axonframework.org/download.

This page offers a number of downloads. Typically, you would want to use the latest stable release. However, if you're eager to get started using the latest and greatest features, you could consider using the snapshot releases instead. The downloads page contains a number of assemblies for you to download. Some of them only provide the Axon library itself, while others also provide the libraries that Axon depends on. There is also a "full" zip file, which contains Axon, its dependencies, the sources and the documentation, all in a single download.

If you really want to stay on the bleeding edge of development, you can clone the Git repository: git://github.com/AxonFramework/AxonFramework.git, or visit https://github.com/AxonFramework/AxonFramework to browse the sources online.

If you use maven as your build tool, you need to configure the correct dependencies for your project. Add the following code in your dependencies section:

<dependency>
    <groupId>org.axonframework</groupId>
    <artifactId>axon-core</artifactId>
    <version>2.0</version>
</dependency>

Most of the features provided by the Axon Framework are optional and require additional dependencies. We have chosen not to add these dependencies by default, as they would potentially clutter your project with artifacts you don't need.

Axon Framework doesn't impose many requirements on the infrastructure. It has been built and tested against Java 6, making that more or less the only requirement.

Since Axon doesn't create any connections or threads by itself, it is safe to run on an Application Server. Axon abstracts all asynchronous behavior by using Executors, meaning that you can easily pass a container managed Thread Pool, for example. If you don't use an Application Server (e.g. Tomcat, Jetty or a stand-alone app), you can use the Executors class or the Spring Framework to create and configure Thread Pools.

While implementing your application, you might run into problems, wonder about why certain things are the way they are, or have some questions that need an answer. The Axon Users mailinglist is there to help. Just send an email to axonframework@googlegroups.com. Other users as well as contributors to the Axon Framework are there to help with your issues.

If you find a bug, you can report them at axonframework.org/issues. When reporting an issue, please make sure you clearly describe the problem. Explain what you did, what the result was and what you expected to happen instead. If possible, please provide a very simple Unit Test (JUnit) that shows the problem. That makes fixing it a lot simpler.

Axon's main modules are the modules that have been thorouhgly tested and are robust enough to use in demanding production environments. The maven groupId of all these modules is org.axonframework.

Incubator modules have not undergone the same amount of testing as the main modules, are not as well documented, and may therefore not be suitable for demanding production environments. They are typically work-in-progress and may have some "rough edges". Use these modules at your own peril.

The maven groupId for incubator modules is org.axonframework.incubator.

Both your custom gateway and the one provided by Axon need to be configured with at least access to the Command Bus. In addition, the Command Gateway can be configured with a RetryScheduler and any number of CommandDispatchInterceptors.

The RetryScheduler is capable of scheduling retries when command execution has failed. The IntervalRetryScheduler is an implementation that will retry a given command at set intervals until it succeeds, or a maximum number of retries is done. When a command fails due to an exception that is explicitly non-transient, no retries are done at all. Note that the retry scheduler is only invoked when a command fails due to a RuntimeException. Checked exceptions are regarded "business exception" and will never trigger a retry.

CommandDispatchInterceptors allow modification of CommandMessages prior to dispatching them to the Command Bus. In contrast to CommandDispatchInterceptors configured on the CommandBus, these interceptors are only invoked when messages are sent through this gateway. The interceptors can be used to attach meta data to a command or do validation, for example.

The GatewayProxyFactory creates an instance of a Command Gateway based on an interface class. The behavior of each method is based on the parameter types, return type and declared exception. Using this gateway is not only convenient, it makes testing a lot easier by allowing you to mock your interface where needed.

This is how parameter affect the behavior of the CommandGateway:

The declared return value of a method will also affect its behavior:

Exceptions have the following effect:

Finaly, there is the possiblity to use annotations:

public interface MyGateway {
    
    // fire and forget
    void sendCommand(MyPayloadType command);

    // method that attaches meta data and will wait for a result for 10 seconds
    @Timeout(value = 10, unit = TimeUnit.SECONDS)
    ReturnValue sendCommandAndWaitForAResult(MyPayloadType command, 
                                             @MetaData("userId") String userId);

    // alternative that throws exceptions on timeout
    @Timeout(value = 20, unit = TimeUnit.SECONDS)
    ReturnValue sendCommandAndWaitForAResult(MyPayloadType command) 
                         throws TimeoutException, InterruptedException;

    // this method will also wait, caller decides how long
    void sendCommandAndWait(MyPayloadType command, long timeout, TimeUnit unit) 
                         throws TimeoutException, InterruptedException;
}

The CommandBus provides two methods to dispatch commands to their respective handler: dispatch(commandMessage, callback) and dispatch(commandMessage). The first parameter is a message containing the actual command to dispatch. The optional second parameter takes a callback that allows the dispatching component to be notified when command handling is completed. This callback has two methods: onSuccess() and onFailure(), which are called when command handling returned normally, or when it threw an exception, respectively.

The calling component may not assume that the callback is invoked in the same thread that dispatched the command. If the calling thread depends on the result before continuing, you can use the FutureCallback. It is a combination of a Future (as defined in the java.concurrent package) and Axon's CommandCallback. Alternatively, consider using a Command Gateway.

Best scalability is achieved when your application is not interested in the result of a dispatched command at all. In that case, you should use the single-parameter version of the dispatch method. If the CommandBus is fully asynchronous, it will return immediately after the command has been successfully dispatched. Your application will just have to guarantee that the command is processed and with "positive outcome", sooner or later...

The SimpleCommandBus is, as the name suggests, the simplest implementation. It does straightforward processing of commands in the thread that dispatches them. After a command is processed, the modified aggregate(s) are saved and generated events are published in that same thread. In most scenario's, such as web applications, this implementation will suit your needs.

The SimpleCommandBus allows interceptors to be configured. CommandDispatchInterceptors are invoked when a command is dispatched on the Command Bus. The CommandHandlerInterceptors are invoked before the actual command handler method is, allowing you to do modify or block the command. See Section 3.5, “Command Interceptors” for more information.

The SimpleCommandBus maintains a Unit of Work for each command published. This Unit of Work is created by a factory: the UnitOfWorkFactory. To suit any specifc needs your application might have, you can supply your own factory to change the Unit of Work implementation used.

Since all command processing is done in the same thread, this implementation is limited to the JVM's boundaries. The performance of this implementation is good, but not extraordinary. To cross JVM boundaries, or to get the most out of your CPU cycles, check out the other CommandBus implementations.

The SimpleCommandBus has reasonable performance characteristics, especially when you've gone through the performance tips in Chapter 10, Performance Tuning. The fact that the SimpleCommandBus needs locking to prevent multiple threads from concurrently accessing the same aggregate causes processing overhead and lock contention.

The DisruptorCommandBus takes a different approach to multithreaded processing. Instead of having multiple threads each doing the same process, there are multiple threads, each taking care of a piece of the process. The DisruptorCommandBus uses the Disruptor (http://code.google.com/p/disruptor), a small framework for concurrent programming, to achieve much better performance, by just taking a different approach to multithreading. Instead of doing the processing in the calling thread, the tasks are handed of to two groups of threads, that each take care of a part of the processing. The first group of threads will execute the command handler, changing an aggregate's state. The second group will store and publish the events to the Event Store and Event Bus.

While the DisruptorCommandBus easily outperforms the SimpleCommandBus by a factor 4(!), there are a few limitations:

A Command Handler must implement the CommandHandler interface. This interface declares only a single method: Object handle(CommandMessage<T> command, UnitOfWork uow), where T is the type of Command this Handler can process. The concept of the UnitOfWork is explained in Section 3.4, “Unit of Work”. Be weary when using return values. Typically, it is a bad idea to use return values to return server-generated identifiers. Consider using client-generated (random) identifiers, such as UUIDs. They allow for a "fire and forget" style of command handlers, where a client does not have to wait for a response. As return value in such a case, you are recommended to simply return null.

You can subscribe and unsubscribe command handlers using the subscribe and unsubscribe methods on CommandBus, respectively. They both take two parameters: the type of command to (un)subscribe the handler to, and the handler to (un)subscribe. An unsubscription will only be successful if the handler passed as the second parameter was currently assigned to handle that type of command. If another command was subscribed to that type of command, nothing happens.

CommandBus commandBus = new SimpleCommandBus();

// to subscribe a handler:
commandBus.subscribe(MyPayloadType.class.getName(), myCommandHandler);

// we can subscribe the same handler to different command types
commandBus.subscribe(MyOtherPayload.class.getName(), myCommandHandler);

// we can also unsubscribe the handler from one of these types:
commandBus.unsubscribe(MyOtherPayload.class.getName(), myCommandHandler);

// we don't have to use the payload to identifier the command type (but it's a good default)
commandBus.subscribe("MyCustomCommand", myCommandHandler);

More often than not, a command handler will need to process several types of closely related commands. With Axon's annotation support you can use any POJO as command handler. Just add the @CommandHandler annotation to your methods to turn them into a command handler. These methods should declare the command to process as the first parameter. They may take optional extra parameters, such as the UnitOfWork for that command (see Section 3.4, “Unit of Work”). Note that for each command type, there may only be one handler! This restriction counts for all handlers registered to the same command bus.

	Note
	If you use Spring, you can add the <axon:annotation-config/> element to your application context. It will turn any bean with @CommandHandler annotated methods into a command handler. They will also be automatically subscribed to the CommandBus. In combination with Spring's classpath scanning (@Component), this will automatically subscribe any command handler in your application.

AggregateAnnotationCommandHandler

It is not unlikely that most command handler operations have an identical structure: they load an Aggregate from a repository and call a method on the returned aggregate using values from the command as parameter. If that is the case, you might benefit from a generic command handler: the AggregateAnnotationCommandHandler. This command handler uses @CommandHandler annotations on the aggregate's methods to identify which methods need to be invoked for an incoming command. If the @CommandHandler annotation is placed on a constructor, that command will cause a new Aggregate instance to be created.

The AggregateAnnotationCommandHandler still needs to know which aggregate instance (identified by it's unique Aggregate Identifier) to load and which version to expect. By default, the AggregateAnnotationCommandHandler uses annotations on the command object to find this information. The @TargetAggregateIdentifier annotation must be put on a field or getter method to indicate where the identifier of the target Aggregate can be found. Similarly, the @TargetAggregateVersion may be used to indicate the expected version.

The @TargetAggregateIdentifier annotation can be placed on a field or a method. The latter case will use the return value of a method invocation (without parameters) as the value to use.

If you prefer not to use annotations, the behavior can be overridden by supplying a custom CommandTargetResolver. This class should return the Aggregate Identifier and expected version (if any) based on a given command.

Creating new Aggregate Instances

When the @CommandHandler annotation is placed on an Aggregate's constructor, the respective command will create a new instance of that aggregte and add it to the repository. Those commands do not require to target a specific aggregate instance. That wouldn't make sense, since the instance is yet to be created. Therefore, those commands do not require any @TargetAggregateIdentifier or @TargetAggregateVersion annotations, nor will a custom CommandTargetResolver be invoked for these commands.

public class MyAggregate extends AbstractAnnotatedAggregateRoot {
   
    @AggregateIdentifier
    private String id;

    @CommandHandler
    public MyAggregate(CreateMyAggregateCommand command) {
        apply(new MyAggregateCreatedEvent(IdentifierFactory.getInstance().generateIdentifier()));
    }

    // no-arg constructor for Axon
    MyAggregate() {
    }

    @CommandHandler
    public void doSomething(DoSomethingCommand command) {
        // do something...
    }

    // code omitted for brevity. The event handler for MyAggregateCreatedEvent must set the id field
}

public class DoSomethingCommand {

    @TargetAggregateIdentifier
    private String aggregateId;

    // code omitted for brevity

}

// to generate the command handlers for this aggregate:
AggregateAnnotationCommandHandler handler = AggregateAnnotationCommandHandler.subscribe(MyAggregate.class, repository, commandBus);
// or when using another type of CommandTargetResolver:
AggregateAnnotationCommandHandler handler = AggregateAnnotationCommandHandler.subscribe(MyAggregate.class, repository, commandBus, myOwnCommandTargetResolver);

// to unsubscribe:
handler.unsubscribe();

Command Dispatch Interceptors are invoked when a command is dispatched on a Command Bus. They have the ability to alter the Command Message, by adding Meta Data, for example, or block the command by throwing an Exception. These interceptors are always invoked on the thread that dispatches the Command.

3.5.1.1. Structural validation

There is no point in processing a command if it does not contain all required information in the correct format. In fact, a command that lacks information should be blocked as early as possible, preferably even before any transaction is started. Therefore, an interceptor should check all incoming commands for the availability of such information. This is called structural validation.

Axon Framework has support for JSR 303 Bean Validation based validation. This allows you to annotate the fields on commands with annotations like @NotEmpty and @Pattern. You need to include a JSR 303 implementation (such as Hibernate-Validator) on your classpath. Then, configure a BeanValidationInterceptor on your Command Bus, and it will automatically find and configure your validator implementation. While it uses sensible defaults, you can fine-tune it to your specific needs.

	Tip
	You want to spend as less resources on an invalid command as possible. Therefore, this interceptor is generally placed in the very front of the interceptor chain. In some cases, a Logging or Auditing interceptor might need to be placed in front, with the validating interceptor immediately following it.

The BeanValidationInterceptor also implements CommandHandlerInterceptor, allowing you to configure it as a Handler Interceptor as well.

Command Handler Interceptors can take action both before and after command processing. Interceptors can even block command processing altogether, for example for security reasons.

Interceptors must implement the CommandHandlerInterceptor interface. This interface declares one method, handle, that takes three parameters: the command message, the current UnitOfWork and an InterceptorChain. The InterceptorChain is used to continue the dispatching process.

3.5.2.1. Auditing

Well designed events will give clear insight in what has happened, when and why. To use the event store as an Audit Trail, which provides insight in the exact history of changes in the system, this information might not be enough. In some cases, you might want to know which user caused the change, using what command, from which machine, etc.

The AuditingInterceptor is an interceptor that allows you to attach arbitrary information to events just before they are stored or published. The AuditingInterceptor uses an AuditingDataProvider to retrieve the information to attach to these events. You need to provide the implementation of the AuditingDataProvider yourself.

An Audit Logger may be configured to write to an audit log. To do so, you can implement the AuditLogger interface and configure it in the AuditingInterceptor. The audit logger is notified both on succesful execution of the command, as well as when execution fails. If you use event sourcing, you should be aware that the event log already contains the exact details of each event. In that case, it could suffice to just log the event identifier or aggregate identifier and sequence number combination.

	Note
	Note that the log method is called in the same thread as the command processing. This means that logging to slow sources may result in higher response times for the client. When important, make sure logging is done asynchronously from the command handling thread.

The JGroupsConnector uses (as the name already gives away) JGroups as the underlying discovery and dispatching mechanism. Describing the feature set of JGroups is a bit too much for this reference guide, so please refer to the JGroups User Guide for more details.

The JGroupsConnector has four mandatory configuration elements:

Ultimately, the JGroupsConnector needs to actually connect, in order to dispatch Messages to other segments. To do so, call the connect() method. It takes a single parameter: the load factor. The load factor defines how much load, relative to the other segments this segment should receive. A segment with twice the load factor of another segment will be assigned (approximately) twice the amount of routing keys as the other segments. Note that when commands are unevenly distributed over the rouing keys, segments with lower load factors could still receive more command than a segment with a higher load factor.

JChannel channel = new JChannel("path/to/channel/config.xml");
CommandBus localSegment = new SimpleCommandBus();
Serializer serializer = new XStreamSerializer();

JGroupsConnector connector = new JGroupsConnector(channel, "myCommandBus", localSegment, serializer);
DistributedCommandBus commandBus = new DistributedCommandBus(connector);

// on one node:
connector.connect(50);
commandBus.subscribe(CommandType.class.getName(), handler);

// on another node with more CPU:
connector.connect(150);
commandBus.subscribe(CommandType.class.getName(), handler);
commandBus.subscribe(AnotherCommandType.class.getName(), handler2);

// from now on, just deal with commandBus as if it is local...

	Note
	Note that it is not required that all segments have Command Handlers for the same type of Comands. You may use different segments for different Command Types altogether. The Distributed Command Bus will always choose a node to dispatch a Command to that has support for that specific type of Command.

Axon framework provides a few repository implementations that can use event sourcing as storage method for aggregates. These repositories require that aggregates implement the EventSourcedAggregateRoot interface. As with most interfaces in Axon, we also provide one or more abstract implementations to help you on your way.

AbstractEventSourcedAggregateRoot

The AbstractEventSourcedAggregateRoot implements all methods on the EventSourcedAggregateRoot interface. It defines an abstract handle() method, which you need to implement with the actual logic to apply state changes based on domain events. When you extend the AbstractEventSourcedAggregateRoot, you can register new events using apply(). This method will register the event to be committed when the aggregate is saved, and will call the handle() method with the event as parameter. You need to implement this handle() method to apply the state changes represented by that event. Below is a sample implementation of an aggregate.

public class MyAggregateRoot extends AbstractEventSourcedAggregateRoot {

    private String aggregateIdentifier;
    private String someProperty;

    public MyAggregateRoot(String id) {
        apply(new MyAggregateCreatedEvent(id));
    }

    // constructor required for reconstruction
    protected MyAggregateRoot() {
    }

    public void handle(DomainEvent event) {
        if (MyAggregateCreatedEvent.class.isAssignableFrom(event.getPayloadType())) {
            // make sure to always initialize the aggregate identifier
            this.aggregateIdentifier = ((MyAggregateCreatedEvent) this.getPayload()).getMyIdentifier();
            // do something with someProperty
        }
        // and more if-else-if logic here
    }

    public Object getAggregateIdentifier() {
        return aggregateIdentifier;
    }
}                

isAssignableFrom versus instanceof

Note that the code sample used isAssignableFrom to verify the payload type of a command. This method is preferred over "getPayload() instanceof" for performance reasons. When reading events from the Event Store, they are actually not deserialized until the getPayload() method is invoked. By using getPayloadType() to detect interesting Events, you skip deserialization for events that you are not interested in.

AbstractAnnotatedAggregateRoot

As you see in the example above, the implementation of the handle() method can become quite verbose and hard to read. The AbstractAnnotatedAggregateRoot can help. The AbstractAnnotatedAggregateRoot is a specialization of the AbstractAggregateRoot that provides @EventHandler annotation support to your aggregate. Instead of a single handle() method, you can split the logic in separate methods, with names that you may define yourself. Just annotate the event handler methods with @EventHandler, and the AbstractAnnotatedAggregateRoot will invoke the right method for you.

	Note
	Note that @EventHandler annotated methods on an AbstractAnnotatedAggregateRoot are only called when events are applied directly to the aggregate locally. This should not be confused with annotating event handler methods on EventListener classes, in which case event handler methods handle events dispatched by the EventBus. See Section 6.2, “Event Listeners”.

public class MyAggregateRoot extends AbstractAnnotatedAggregateRoot {

    @AggregateIdentifier
    private String aggregateIdentifier;
    private String someProperty;

    public MyAggregateRoot(String id) {
        apply(new MyAggregateCreatedEvent(id));
    }

    // constructor needed for reconstruction
    protected MyAggregateRoot() {
    }

    @EventHandler
    private void handleMyAggregateCreatedEvent(MyAggregateCreatedEvent event) {
        // make sure identifier is always initialized properly
        this.aggregateIdentifier = event.getMyAggregateIdentifier();
        // do something with someProperty
    }
}                

@EventHandler annotated methods are resolved using specific rules. These rules are thorougly explained in Section 6.2.2, “Annotated Event Handler”.

Using private methods

Event handler methods may be private, as long as the security settings of the JVM allow the Axon Framework to change the accessibility of the method. This allows you to clearly separate the public API of your aggregate, which exposes the methods that generate events, from the internal logic, which processes the events. Most IDE's have an option to ignore "unused private method" warnings for methods with a specific annotation. Alternatively, you can add an @SuppressWarnings("UnusedDeclaration") annotation to the method.

The AbstractAnnotatedAggregateRoot also requires that the field containing the aggregate identifier is annotated with @AggregateIdentifier. If you use JPA and have JPA annotations on the aggregate, Axon can also use the @Id annotation provided by JPA.

Complex business logic often requires more than what an aggregate with only an aggregate root can provide. In that case, it important that the complexity is spread over a number of entities within the aggregate. When using event sourcing, not only the aggregate root needs to use event to trigger state transitions, but so does each of the entities within that aggregate.

Axon provides support for event sourcing in complex aggregate structures. All entities other than the aggregate root need to implement EventSourcedEntity (but extending AbstractEventSourcedEntity or AbstractAnnotatedEntity makes implementing these entities easier). The EventSourcedAggregateRoot implementations provided by Axon Framework are aware of these entities and will call their event handlers when needed.

When an entity (including the aggregate root) applies an Event, it is registered with the Aggregate Root. The aggregate root applies the event locally first. Next, it will evaluate all fields annotated with @EventSourcedMember for any implementations of EventSourcedEntity and handle the event on them. Each entity does the same thing to its fields.

To register an Event, the Entity must know about the Aggregate Root. Axon will automatically register the Aggregate Root with an Entity before applying any Events to it. This means that Entities (in constrast to the Aggregate Root) should never apply an Event in their constructor. Non-Aggregate Root Entities should be created in an @EventHandler annotated method in their parent Entity (as creation of an entity can be considered a state change of the Aggregate). Axon will ensure that the Aggregate Root is properly registered in time. This also means that a single Entity cannot be part of more than one aggregate at any time.

Fields that (may) contain child entities must be annotated with @EventSourcedMember. This annotation may be used on a number of field types:

If the value contained in an @EventSourcedMember annotated field does not implement EventSourcedEntity, it is simply ignored.

If you need to reference an Entity from any other location than the above mentioned, you can override the getChildEntities() method. This method should return a Collection of entities that should be notified of the Event. Note that each entity is invoked once for each time it is located in the returned Collection.

Note that in high-performance situations, the reflective approach to finding child entities can be costly. In that case, you can override the getChildEntities() method.

The FileSystemEventStore stores the events in a file on the file system. It provides good performance and easy configuration. The downside of this event store is that is does not provide transaction support and doesn't cluster very well. The only configuration needed is the location where the event store may store its files and the serializer to use to actually serialize and deserialize the events.

Note that the FileSystemEventStore is not aware of transactions and cannot automatically recover from crashes. Furthermore, it stores a single file for each aggregate, potentially creating too many files for the OS to handle. It is therefore not a suitable implementation for production environments.

Out of diskspace

When using the FileSystemEventStore in a test environment (or other environment where many aggregate may be created), you may end up with an "Out of disk space" error, even if there is plenty of room on the disk. The cause is that the filesystem runs out of i-nodes (or an equivalent when not on a Unix filesystem). This typically means that a filesystem holds too many files. To prevent this problem, make sure the output directory of the FileSystemEventStore is cleaned after each test run.

The JpaEventStore stores events in a JPA-compatible data source. Unlike the file system version, the JPAEventStore supports transactions. The JPA Event Store stores events in so called entries. These entries contain the serialized form of an event, as well as some fields where meta-data is stored for fast lookup of these entries. To use the JpaEventStore, you must have the JPA (javax.persistence) annotations on your classpath.

By default, the event store needs you to configure your persistence context (defined in META-INF/persistence.xml file) to contain the classes DomainEventEntry and SnapshotEventEntry (both in the org.axonframework.eventstore.jpa package).

Below is an example configuration of a persistence context configuration:

<persistence xmlns="http://java.sun.com/xml/ns/persistence" version="1.0">
    <persistence-unit name="eventStore" transaction-type="RESOURCE_LOCAL">
        <class>org...eventstore.jpa.DomainEventEntry</class> 
        <class>org...eventstore.jpa.SnapshotEventEntry</class>
    </persistence-unit>
</persistence>

Detecting duplicate key violations in the database

Axon uses Locking to prevent two threads from accessing the same Aggregate. However, if you have multiple JVMs on the same database, this won't help you. In that case, you'd have to rely on the database to detect conflicts. Concurrent access to the event store will result in a Key Constraint Violation, as the table only allows a single Event for an aggregate with any sequence number. Inserting a second event for an existing aggregate with an existing sequence number will result in an error. The JPA EventStore can detect this error and translate it to a ConcurrencyException. However, each database system reports this violation differently. If you register your DataSource with the JpaEventStore, it will try to detect the type of database and figure out which error codes represent a Key Constraint Violation. Alternatively, you may provide a PersistenceExceptionTranslator instance, which can tell if a given exception represents a Key Constraint Violation. If no DataSource or PersistenceExceptionTranslator is provided, exceptions from the Database driver are thrown as-is.

public class MyEntityManagerProvider implements EntityManagerProvider {

    private EntityManager entityManager;

    @Override
    public EntityManager getEntityManager() {
        return entityManager;
    }

    @PersistenceContext(unitName = "myPersistenceUnit")
    public void setEntityManager(EntityManager entityManager) {
        this.entityManager = entityManager;
    }                

Customizing the Event storage

By default, the JPA Event Store stores entries in DomainEventEntry and SnapshotEventEntry entities. While this will suffice in many cases, you might encounter a situation where the meta-data provided by these entities is not enough. Or you might want to store events of different aggregate types in different tables.

If that is the case, you may provide your own implementation of EventEntryStore in the JPA Event Store's constructor. You will need to provide implementations of methods that load and store serialized events. Check the API Documentation of the EventEntryStore class for implementation requirements.

Memory consumption warning

Note that persistence providers, such as Hibernate, use a first-level cache on their EntityManager implementation. Typically, this means that all entities used or returned in queries are attached to the EntityManager. They are only cleared when the surrounding transaction is committed or an explicit "clear" in performed inside the transaction. This is especially the case when the Queries are executed in the context of a transaction. To work around this issue, make sure to exclusively query for non-entity objects. You can use JPA's "SELECT new SomeClass(parameters) FROM ..." style queries to work around this issue. Alternatively, call EntityManager.flush() and EntityManager.clear() after fetching a batch of events. Failure to do so might result in OutOfMemoryExceptions when loading large streams of events.

	Hibernate support
	The EventEntryStore implementation used by Axon automatically detects whether Hibernate is used as JPA implementation. If that is the case, it will use some Hiberante-specific features to optimize performance while reducing memory consumption. To prevent Hibernate specific logic, configure the JPA Event Store to use the DefaultEventEntryStore instead.

MongoDB is a document based NoSQL store. It scalability characteristics make it suitable for use as an Event Store. Axon provides the MongoEventStore, which uses MongoDB as backing database. It is contained in the Axon Mongo module (Maven artifactId axon-mongo).

Events are stored in two separate collections: one for the actual event streams and one for the snapshots.

By default, the MongoEventStore stores each event in a separate document. It is, however, possible to change the StorageStrategy used. The alternative provided by Axon is the DocumentPerCommitStorageStrategy, which creates a single document for all Events that have been stored in a single commit (i.e. in the same DomainEventStream).

Storing an entire commit in a single document has the advantage that a commit is stored atomically. Furthermore, it requires only a single roundtrip for any number of events. A disadvantage is that it becomes harder to query events manually or through the EventStoreManagement methods. When refactoring the domain model, for example, it is harder to "transfer" events from one aggregate to another if they are included in a "commit document".

The MongoDB doesn't take a lot of configuration. All it needs is a reference to the collections to store the Events in, and you're set to go.

If you have specific requirements for an event store, it is quite easy to implement one using different underlying data sources. Reading and appending events is done using a DomainEventStream, which is quite similar to iterator implementations.

Instead of eagerly deserializing Events, consider using the SerializedDomainEventMessage, which will postpone deserialization of Meta Data and Payload until it is actually used by a handler.

	Tip
	The SimpleDomainEventStream class will make the contents of a sequence ( List or array) of EventMessage instances accessible as event stream.

Event Stores need a way to serialize the Event to prepare it for storage. By default, Axon provides the XStreamSerializer, which uses XStream to serialize Events into XML and vice versa. XStream is reasonably fast and is more flexible than Java Serialization. Furthermore, the result of XStream serialization is human readable. Quite useful for logging and debugging purposes.

The XStreamSerializer can be configured. You can define aliases it should use for certain packages, classes or even fields. Besides being a nice way to shorten potentially long names, aliases can also be used when class definitions of event change. For more information about aliases, visit the XStream website.

	Spring XML Configuration and Serializer Customization
	Configuring the serializer using Java code (or other JVM languages) is easy. However, configuring it in a Spring XML Application Context is not so trivial, due to its limitations to invoke methods. One of the options is to create a FactoryBean that creates an instance of an XStreamSerializer and configures it in code. Check the Spring Reference

You may also implement your own Serializer, simply by creating a class that implements Serializer, and configuring the Event Store to use that implementation instead of the default.

To explain how to write an upcaster for Axon we'll walk through a small example, describing the details of writing an upcaster as we go along.

Let's assume that there is an Event Store containing many AdministrativeDetailsUpdated events. New requirements have let to the introduction of two new events: AddressUpdatedEvent and InsurancePolicyUpdatedEvent. Previously though, all information in these two events was contained in the old AdministrativeDetailsUpdatedEvent, which is now deprecated. To nicely handle this situation we'll write an upcaster to transform the AdministrativeDetailsUpdatedEvent into an AddressUpdatedEvent and an InsurancePolicyUpdatedEvent.

Here is the code for an upcaster:

import org.dom4j.Document;
                
public class AdministrativeDetailsUpdatedUpcaster implements Upcaster<Document> { 

    @Override
    public boolean canUpcast(SerializedType serializedType) { 
        return serializedType.getName().equals("org.example.AdministrativeDetailsUpdated") &&
               "0".equals(serializedType.getRevision());
    }

    @Override
    public Class<Document> expectedRepresentationType() { 
        return Document.class;
    }

    @Override
    public List<SerializedObject<Document>> upcast(SerializedObject<Document> intermediateRepresentation,
                                                   List<SerializedType> expectedTypes, UpcastingContext context) {
        Document administrativeDetailsUpdatedEvent = intermediateRepresentation.getData();

        Document addressUpdatedEvent =
                new DOMDocument(new DOMElement("org.example.InsurancePolicyUpdatedEvent"));
        addressUpdatedEvent.getRootElement()
                .add(administrativeDetailsUpdatedEvent.getRootElement().element("address"));

        Document insurancePolicyUpdatedEvent =
                new DOMDocument(new DOMElement("org.example.InsurancePolicyUpdatedEvent"));
        insurancePolicyUpdatedEvent.getRootElement()
                .add(administrativeDetailsUpdatedEvent.getRootElement().element("policy"));

        List<SerializedObject<?>> upcastedEvents = new ArrayList<SerializedObject<?>>();
        upcastedEvents.add(new SimpleSerializedObject<Document>(
                addressUpdatedEvent, Document.class, expectedTypes.get(0)));
        upcastedEvents.add(new SimpleSerializedObject>Document>(
                insurancePolicyUpdatedEvent, Document.class, expectedTypes.get(1)));
        return upcastedEvents;
    }

    @Override
    public List<SerializedType> upcast(SerializedType serializedType) { 
        SerializedType addressUpdatedEventType = 
                new SimpleSerializedType("org.example.AddressUpdatedEvent", "1");
        SerializedType insurancePolicyUpdatedEventType = 
                new SimpleSerializedType("org.example.InsurancePolicyUpdatedEvent", "1");
        return Arrays.asList(addressUpdatedEventType, insurancePolicyUpdatedEventType);
    }
}

The Upcaster Chain is responsible for upcasting events by chaining the output of one upcaster to the next. It comes in the following two flavours:

The LazyUpcasterChain is a safe choice if your upcasters are stateless or do not depend on other upcasters. Always consider using the LazyUpcasterChain since it can provide a great performance benefit over the SimpleUpcasterChain. If you want guaranteed upcasting in a strict order, use the SimpleUpcasterChain.

An upcaster works on a given content type (e.g. dom4j Document). To provide extra flexibility between upcasters, content types between chained upcasters may vary. Axon will try to convert between the content types automatically by using ContentTypeConverters. It will search for the shortest path from type x to type y, perform the conversion and pass the converted value into the requested upcaster. For performance reasons, conversion will only be performed if the canUpcast method on the receiving upcaster yields true.

The ContentTypeConverters may depend on the type of serializer used. Attempting to convert a byte[] to a dom4j Document will not make any sence unless a Serializer was used that writes an event as XML. To make sure the UpcasterChain has access to the serializer-specific ContentTypeConverters, you can pass a reference to the serializer to the constructor of the UpcasterChain.

	Tip
	To achieve the best performance, ensure that all upcasters in the same chain (where one's output is another's input) work on the same content type.

If the content type conversion that you need is not provided by Axon you can always write one yourself using the ContentTypeConverter interface.

Snapshot creation can be triggered by a number of factors, for example the number of events created since the last snapshot, the time to initialize an aggregate exceeds a certain threshold, time-based, etc. Currently, Axon provides a mechanism that allows you to trigger snapshots based on an event count threshold.

The EventCountSnapshotterTrigger provides the mechanism to trigger snapshot creation when the number of events needed to load an aggregate exceeds a certain threshold. If the number of events needed to load an aggregate exceeds a certain configurable threshold, the trigger tells a Snapshotter to create a snapshot for the aggregate.

The snapshot trigger is configured on an Event Sourcing Repository and has a number of properties that allow you to tweak triggering:

A Snapshotter is responsible for the actual creation of a snapshot. Typically, snapshotting is a process that should disturb the operational processes as little as possible. Therefore, it is recommended to run the snapshotter in a different thread. The Snapshotter interface declares a single method: scheduleSnapshot(), which takes the aggregate's type and identifier as parameters.

Axon provides the AggregateSnapshotter, which creates and stores AggregateSnapshot instances. This is a special type of snapshot, since it contains the actual aggregate instance within it. The repositories provided by Axon are aware of this type of snapshot, and will extract the aggregate from it, instead of instantiating a new one. All events loaded after the snapshot events are streamed to the extracted aggregate instance.

	Note
	Do make sure that the Serializer instance you use (which defaults to the XStreamSerializer) is capable of serializing your aggregate. The XStreamSerializer requires you to use either a Hotspot JVM, or your aggregate must either have an accessible default constructor or implement the Serializable interface.

The AbstractSnapshotter provides a basic set of properties that allow you to tweak the way snapshots are created:

The AggregateSnapshotter provides on more property:

Note

If you use an executor that executes snapshot creation in another thread, make sure you configure the correct transaction management for your underlying event store, if necessary. Spring users can use the SpringAggregateSnapshotter, which allows you to configure a PlatformTransactionManager. The SpringAggregateSnapshotter will autowire all aggregate factories (either directly, or via the Repository), if a list is not explicitly configured.

All Axon-provided Event Store implementations are capable of storing snapshot events. They provide a special method that allows a DomainEventMessage to be stored as a snapshot event. You have to initialize the snapshot event completely, including the aggregate identifier and the sequence number. There is a special constructor on the GenericDomainEventMessage for this purpose. The sequence number must be equal to the sequence number of the last event that was included in the state that the snapshot represents. In most cases, you can use the getVersion() on the AggregateRoot (which each event sourced aggregate implements) to obtain the sequence number to use in the snapshot event.

When a snapshot is stored in the Event Store, it will automatically use that snapshot to summarize all prior events and return it in their place. All event store implementations allow for concurrent creation of snapshots. This means they allow snapshots to be stored while another process is adding Events for the same aggregate. This allows the snapshotting process to run as a separate process alltogether.

	Note
	Normally, you can archive all events once they are part of a snapshot event. Snapshotted events will never be read in again by the event store in regular operational scenario's. However, if you want to be able to reconstruct aggregate state prior to the moment the snapshot was created, you must keep the events up to that date.

Axon provides a special type of snapshot event: the AggregateSnapshot, which stores an entire aggregate as a snapshot. The motivation is simple: your aggregate should only contain the state relevant to take business decisions. This is exactly the information you want captured in a snapshot. All Event Sourcing Repositories provided by Axon recognize the AggregateSnapshot, and will extract the aggregate from it. Beware that using this snapshot event requires that the event serialization mechanism needs to be able to serialize the aggregate.

A snapshot event is an event like any other. That means a snapshot event is handled just like any other domain event. When using annotations to demarcate event handers (@EventHandler), you can annotate a method that initializes full aggregate state based on a snapshot event. The code sample below shows how snapshot events are treated like any other domain event within the aggregate.

public class MyAggregate extends AbstractAnnotatedAggregateRoot {

    // ... code omitted for brevity

    @EventHandler
    protected void handleSomeStateChangeEvent(MyDomainEvent event) {
        // ...
    }

    @EventHandler
    protected void applySnapshot(MySnapshotEvent event) {
        // the snapshot event should contain all relevant state
        this.someState = event.someState;
        this.otherState = event.otherState;
    }
}                

There is one type of snapshot event that is treated differently: the AggregateSnapshot. This type of snapshot event contains the actual aggregate. The aggregate factory recognizes this type of event and extracts the aggregate from the snapshot. Then, all other events are re-applied to the extracted snapshot. That means aggregates never need to be able to deal with AggregateSnapshot instances themselves.

Once a snapshot event is written, it prevents older events and snapshot events from being read. Domain Events are still used in case a snapshot event becomes obsolete due to changes in the structure of an aggregate. The older snapshot events are hardly ever needed. SnapshotEventStore implementation may choose to keep only a limited amount of snapshots (e.g. only one) for each aggregate.

The JpaEventStore allows you to configure the amount of snapshots to keep per aggregate. It defaults to 1, meaning that only the latest snapshot event is kept for each aggregate. Use setMaxSnapshotsArchived(int) to change this setting. Use a negative integer to prevent pruning altogether.

The SimpleEventBus is, as the name suggests, a very basic implementation of the EventBus interface. It just dispatches each incoming Event to each of the subscribed EventListeners sequentially. If an EventListener throws an Exception, dispatching stops and the exception is propagated to the component publising the Event.

The SimpleEventBus is suitable for most cases where dispatching is done synchronously and locally, (i.e. in a single JVM). Once you application requires Events to be published across multiple JVMs, you could consider using the ClusteringEventBus instead.

The ClusteringEventsBus allows application developers to bundle EventListeners into Clusters based on their properties and non-functional requirements. The ClusteringEventBus is also more capable to deal with Events being dispatched among different machines.

Figure 6.1. Structure of the Clustering Event Bus

The ClusteringEventsBus contains two mechanisms: the ClusterSelector, which selects a Cluster instance for each of the registered EventListeners, and the EventBusTerminal, which is responsible for dispatching Events to each of the relevant clusters.

	Background: Axon Terminal
	In the nervous system, an Axon (one of the components of a Neuron) transports electrical signals. These Neurons are interconnected in very complex arrangements. The Axon Terminal is responsible for transmitting these signals from one Neuron to another. More information: www.wikipedia.org/wiki/Axon_terminal.

ClusterSelector

The primary responsibility of the ClusterSelector is to, as the name suggests, select a cluster for each Event Listener that subscribes to the Event Bus. By default, all Event Listeners are placed in a single Cluster instance, which dispatches events to its members sequentially and in the calling thread (similar to how the SimpleEventBus works). By providing a custom implementation, you can arrange the Event Listeners into different Cluster instances to suit the requirements of your architecture.

A number of ClusterSelector implementations are available. The ClassNamePrefixClusterSelector, for example, uses a mapping of package prefixes to decide which cluster is (most) suitable for an Event Listener. Similarly, the ClassNamePatternClusterSelector uses pattern matching to decide whether a given cluster is suitable. You can use the CompositeClusterSelector to combine several cluster selectors into a single one.

The Cluster interface describes the behavior of a Cluster. By adding information in the Meta Data of a Cluster, the selector can provide hints to the Terminal about the expected behavior.

Clusters and Cluster Selectors in Spring

Spring users can define clusters using the <axon:cluster> element. This element allows you to define the selection criteria for Event Handlers in that cluster. These criteria are transformed into cluster selectors and used to assign each Listener to one of the clusters. By default, this creates a cluster that handles events in the publishing thread. To use a cluster with other semantics, you can define a bean inside the <axon:cluster> element that specifies the Cluster implementation to use. The Clusters are automatically detected and connected to the Event Bus in the application context.

Event listeners need to be registered with an event bus (see Section 6.1, “Event Bus”) to be notified of events. The EventListener interface prescribes a single method to be implemented. This method is invoked for each Event Message passed on the Event Bus that it is subscribed to:

public class MyEventListener implements EventListener {

    public void handle(EventMessage message) {
        if (SomeEvent.class.isAssignableFrom(message.getPayloadType) {
            // handle SomeEvent
        } else if (OtherEvent.class.isAssignableFrom(message.getPayloadType) {
            // handle SomeOtherEvent
        }
    }
}

Implementing the EventListener interface can produce a large if-statement and verbose plumbing code. Using annotations to demarcate Event Handler methods is a cleaner alternative.

AnnotationEventListenerAdapter

The AnnotationEventListenerAdapter can wrap any object into an event listener. The adapter will invoke the most appropriate event handler method available. These event handler methods must be annotated with the @EventHandler annotation.

The AnnotationEventListenerAdapter, as well as the AbstractAnnotatedAggregateRoot, use ParameterResolvers to resolve the value that should be passed in the parameters of methods annotated with @EventHandler. By default, Axon provides a number of parameter resolvers that allow you to use the following parameter types:

• The first parameter is always the payload of the Event message

• Parameters annotated with @MetaData will resolve to the Meta Data value with the key as indicated on the annotation. If required is false (default), null is passed when the meta data value is not present. If required is true, the resolver will not match and prevent the method from being invoked when the meta data value is not present.

• Parameters of type MetaData will have the entire MetaData of an EventMessage injected.

• Parameters annotated with @Timestamp and of type org.joda.time.DateTime will resolve to the timestamp of the EventMessage. This is the time at which the Event was generated.

• Parameters assignable to Message will have the entire EventMessage injected (if the message is assignable to that parameter). If the first parameter is of type message, it effectively matches an Event of any type, even if generic parameters would suggest otherwise. Due to type erasure, Axon cannot detect what parameter is expected. In such case, it is best to declare a parameter of the payload type, followed by a parameter of type Message.

• When using Spring and <axon:annotation-config/> is declared, any other parameters will resolve to autowired beans, if exactly one autowirable candidate is available in the application context. This allows you to inject resources directly into @EventHandler annotated methods.

You can configure additional ParameterResolvers by extending the ParameterResolverFactory class and creating a file named /META-INF/service/org.axonframework.common.annotation.ParameterResolverFactory containing the fully qualified name of the implementing class. Alternatively, you can register your implementation at runtime using ParameterResolverFactory.registerFactory(). Make sure to do so before any adapters are created, otherwise handlers may have been initialized without those parameter resolvers.

In all circumstances, exactly one event handler method is invoked per listener instance. Axon will search the most specific method to invoke, in the following order:

1. On the actual instance level of the class hierarchy (as returned by this.getClass()), all annotated methods are evaluated

2. If one or more methods are found of which all parameters can be resolved to a value, the method with the most specific type is chosen and invoked

3. If no methods are found on this level of the class hierarchy, the super type is evaluated the same way

4. When the top level of the hierarchy is reached, and no suitable event handler is found, the event is ignored.

// assume EventB extends EventA 
// and    EventC extends EventB

public class TopListener {

    @EventHandler
    public void handle(EventA event) {
    }

    @EventHandler
    public void handle(EventC event) {
    }
}

public class SubListener extends TopListener {

    @EventHandler
    public void handle(EventB event) {
    }

}

In the example above, the SubListener will receive all instances of EventB as well as EventC (as it extends EventB). In other words, the TopListener will not receive any invocations for EventC at all. Since EventA is not assignable to EventB (it's its superclass), those will be processed by TopListener.

The constructor of the AnnotationEventListenerAdapter takes two parameters: the annotated bean, and the EventBus, to which the listener should subscribe. You can subscribe and unsubscribe the event listener using the subscribe() and unsubscribe() methods on the adapter. Alternatively, you can use AnnotationEventListenerAdapter.subscribe(listener, eventBus) to create and subscribe the listener in one invocation.

	Tip
	If you use Spring, you can automatically wrap all annotated event listeners with an adapter automatically by adding <axon:annotation-config/> to your application context. Axon will automatically find and wrap annotated event listeners in the Application Context with an AnnotationEventListenerAdapter and register them with the Event Bus.

The Spring AMQP Terminal uses the Spring AMQP module to transmit events to an AMQP compatible message broker, such as Rabbit MQ. It also connects local clusters to queues on that message broker.

The axon-amqp namespace (http://www.axonframework.org/schema/amqp) allows you to configure an AMQP Terminal by adding the <axon-amqp:terminal> element to the Spring application context. On this element, you can define different properties for the terminal, as well as a configuration containing defaults to use to connect each cluster to an AMQP Queue.

The example below shows an example configuration for an AMQP Terminal. The default-configuration element specified the defaults for the Clusters if they don't provide their own values.

<axon-amqp:terminal id="terminal" 
                    connection-factory="amqpConnection" 
                    serializer="serializer"
                    exchange-name="AxonEventBusExchange">
    <axon-amqp:default-configuration transaction-manager="transactionManager"
                                     transaction-size="25" prefetch="200"
                                     error-handler="loggingErrorHandler"/>
</axon-amqp:terminal>

<bean id="amqpConnection" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory"/>

The configure the Spring AMQP Terminal "manually", you need to specify a number of beans in your application context:

When a cluster is selected for an Event Listener, it will be registered with the terminal. At that point, the Spring AMQP terminal will check if there is any cluster-specific configuration available. It does so by checking the AMQP.Config MetaData value. If that value is an instance of AMQPConsumerConfiguration (such as SpringAMQPConsumerConfiguration) any settings configured there will override the defaults set on the terminal itself. This allows you to specify different behavior (such as transaction size) for different clusters.

// XML Configuration for a Cluster with AMQPConsumerConfiguration
<axon:cluster id="myDefaultCluster" default="true">
    <axon:meta-data>
        <entry key="AMQP.Config">
            <axon-amqp:configuration transaction-size="20000"/>
        </entry>
    </axon:meta-data>
</axon:cluster>

As a single Saga instance is responsible for managing a single transaction. That means you need to be able to indicate the start and end of a Saga's Life Cycle.

The AbstractAnnotatedSaga allows you to annotate Event Handlers with an annotation (@SagaEventHandler). If a specific Event signifies the start of a transaction, add another annotation to that same method: @StartSaga. This annotation will create a new saga and invoke its event handler method when a matching Event is published.

By default, a new Saga is only started if no suitable existing Saga (of the same type) can be found. You can also force the creation of a new Saga instance by setting the forceNew property on the @StartSaga annotation to true.

Ending a Saga can be done in two ways. If a certain Event always indicates the end of a Saga's life cycle, annotate that Event's handler on the Saga with @EndSaga. The Saga's Life Cycle will be ended after the invocation of the handler. Alternatively, you can call end() from inside the Saga to end the life cycle. This allows you to conditionally end the Saga.

	Note
	If you don't use annotation support, you need to properly configure your Saga Manager (see Section 7.2.1, “Saga Manager” below). To end a Saga's life cycle, make sure the isActive() method of the Saga returns false.

Event Handling in a Saga is quite comparable to that of a regular Event Listener. The same rules for method and parameter resolution are valid here. There is one major difference, though. While there is a single instance of an Event Listener that deals will all incoming events, multiple instances of a Saga may exist, each interested in different Events. For example, a Saga that manages a transaction around an Order with Id "1" will not be interested in Events regarding Order "2", and vice versa.

public class OrderManagementSaga extends AbstractAnnotatedSaga {

    private boolean paid = false;
    private boolean delivered = false;
    private transient CommandGateway commandGateway;

    @StartSaga
    @SagaEventHandler(associationProperty = "orderId")
    public void handle(OrderCreatedEvent event) {
        // client generated identifiers 
        ShippingId shipmentId = createShipmentId();
        InvoiceId invoiceId = createInvoiceId();
        // associate the Saga with these values, before sending the commands 
        associateWith("shipmentId", shipmentId);
        associateWith("invoiceId", invoiceId);
        // send the commands
        commandGateway.send(new PrepareShippingCommand(...));
        commandGateway.send(new CreateInvoiceCommand(...));
    }

    @SagaEventHandler(associationProperty = "shipmentId")
    public void handle(ShippingArrivedEvent event) {
        delivered = true;
        if (paid) {
            end(); 
        }
    }

    @SagaEventHandler(associationProperty = "invoiceId")
    public void handle(InvoicePaidEvent event) {
        paid = true;
        if (delivered) {
            end(); 
        }
    }

    // ...

}

It is easy to make a Saga take action when something happens. After all, there is an Event to notify the Saga. But what if you want your Saga to do something when nothing happens? That's what deadlines are used for. In invoices, that's typically several weeks, while the confirmation of a credit card payment should occur within a few seconds.

In Axon, you can use an EventScheduler to schedule an Event for publication. In the example of an Invoice, you'd expect that invoice to be paid within 30 days. A Saga would, after sending the CreateInvoiceCommand, schedule an InvoicePaymentDeadlineExpiredEvent to be published in 30 days. The EventScheduler returns a ScheduleToken after scheduling an Event. This token can be used to cancel the schedule, for example when a payment of an Invoice has been received.

Axon provides two EventScheduler implementations: a pure Java one and one using Quartz 2 as a backing scheduling mechanism.

Scheduled Events and Transactions

One or more components will be listening for scheduled Events. These components might rely on a Transaction being bound to the Thread that invokes them. Scheduled Events are published by Threads managed by the EventScheduler. To manage transactions on these threads, you can configure a TransactionManager or aUnitOfWorkFactory that creates Transaction Bound Unit of Work.

	Note
	Spring users can use the QuartzEventSchedulerFactoryBean or SimpleEventSchedulerFactoryBean for easier configuration. It allows you to set the PlatformTransactionManager directly.

Sagas generally do more than just maintaining state based on Events. They interact with external components. To do so, they need access to the Resources necessary to address to components. Usually, these resources aren't really part of the Saga's state and shouldn't be persisted as such. But once a Saga is reconstructed, these resources must be injected before an Event is routed to that instance.

For that purpose, there is the ResourceInjector. It is used by the SagaRepository (for existing Saga instances) and the SagaFactory (for newly created instances) to inject resources into a Saga. Axon provides a SpringResourceInjector, which injects annotated fields and methods with Resources from the Application Context, and a SimpleResourceInjector, which detects setters and injects resources which have been registered with it.

	Mark fields holding injected resources transient
	Since resources should not be persisted with the Saga, make sure to add the transient keyword to those fields. This will prevent the serialization mechanism to attempt to write the contents of these fields to the repository. The repository will automatically re-inject the required resources after a Saga has been deserialized.

The SagaManager is responsible for redirecting Events to the appropriate Saga instances and managing their life cycle. There are two SagaManager implementations in Axon Framework: the AnnotatedSagaManager, which provides the annotation support and the SimpleSagaManager, which is less powerful, but doesn't force you into using annotations.

Sagas operate in a highly concurrent environment. Multiple Events may reach a Saga at (nearly) the same time. This means that Sagas need to be thread safe. By default, Axon's SagaManager implementations will synchronize access to a Saga instance. This means that only one thread can access a Saga at a time, and all changes by one thread are guaranteed to be visible to any successive threads (a.k.a happens-before order in the Java Memory Model). Optionally, you may switch this locking off, if you are sure that your Saga is completely thread safe on its own. Just setSynchronizeSagaAccess(false). When disabling synchronization, do take note of the fact that this will allow a Saga to be invoked while it is in the process of being stored by a repository. The result may be that a Saga is stored in an inconsistent state first, and overwritten by it's actual state later.

<axon:saga-manager id="sagaManager" saga-factory="optionalSagaFactory"
                   saga-repository="sagaRepository" event-bus="eventBus">
    <axon:types>
        fully.qualified.ClassName,
        another.fq.ClassName
    </axon:types>
</axon:saga-manager>

Asynchronous Event Handling for Sagas

As with Event Listeners, it is also possible to asynchronously handle events for sagas. To handle events asynchronously, the SagaManager needs to be configured with an Executor implementation. The Executor supplies the threads needed to process the events asynchronously. Often, you'll want to use a thread pool. You may, if you want, share this thread pool with other asynchronous activities.

When an executor is provided, the SagaManager will automatically use it to find associated Saga instances and dispatch the events each of these instances. The SagaManager will guarantee that for each Saga instance, all events are processed in the order they arrive. For optimization purposes, this guarantee does not count in between Sagas.

Because Transactions are often Thread bound, you may need to configure a Transaction Manager with the SagaManager. This transaction manager is invoked before and after each invocation to the Saga Repository and before and after each batch of Events has been processed by the Saga itself.

In a Spring application context, a Saga Manager can be marked as asynchronous by adding the executor and optionally the transaction-manager attributes to the saga-manager element, as shown below. The processor-count attribute defines the number of threads that should process the sagas.

<axon:saga-manager id="sagaManager" saga-factory="optionalSagaFactory"
                   saga-repository="sagaRepository" event-bus="eventBus">
    <axon:async processor-count="10" executor="myThreadPool" transaction-manager="txManager"/>
    <axon:types>
        fully.qualified.ClassName,
        another.fq.ClassName
    </axon:types>
</axon:saga-manager>

The transaction-manager should point to a PlatformTransactionManager, Spring's interface for transaction managers. Generally you can use the same transaction manager as the other components in your application (e.g. JpaTransactionManager).

Exceptions while processing Events

Since the processing is asynchronous, Exceptions cannot be propagated to the components publishing the Event. Exceptions raised by Sagas while handling an Event are logged and discarded. When an exception occurs while persisting Saga state, the SagaManager will retry persisting the Saga at a later stage. If there are incoming Events, they are processed using the in-memory representation, re-attemtping the persistence later on. When shutting down the application, the SagaManager will do a final attempt to persist the Sagas to their repository, and give up if it does not succeed, allowing the application to finish the shutdown process.

The SagaRepository is responsible for storing and retrieving Sagas, for use by the SagaManager. It is capable of retrieving specific Saga instances by their identifier as well as by their Association Values.

There are some special requirements, however. Since concurrency in Sagas is a very delicate procedure, the repository must ensure that for each conceptual Saga instance (with equal identifier) only a single instance exists in the JVM.

Axon provides three SagaRepository implementations: the InMemorySagaRepository, the JpaSagaRepository and the MongoSagaRepository.

If a database backed Saga Repository is used, saving and loading Saga instances may be a relatively expensive operation. Especially in situations where the same Saga instance is invoked multiple times within a short timespan, a cache can be beneficial to the application's performance.

Axon provides the CachingSagaRepository implementation. It is a Saga Repository that wraps another repository, which does the actual storage. When loading Sagas or Association Values, the CachingSagaRepository will first consult its caches, before delegating to the wrapped repository. When storing information, all call are always delegated, to ensure that the backing storage always has a consistent view on the Saga's state.

To configure caching, simply wrap any Saga repository in a CachingSagaRepository. The constructor of the CachingSagaRepository takes three parameters: the repository to wrap and the caches to use for the Association Values and Saga instances, respectively. The latter two arguments may refer to the same cache, or to different ones. This depends on the eviction requirements of your specific application.

<axon:jpa-saga-repository id="cachingSagaRepository">
    <axon:cache-config saga-cache="sagaCacheRef" associations-cache="associationCacheRef"/>
</axon:jpa-saga-repository>

Using the annotated event listeners is very easy when you use Spring. All you need to do is configure the AnnotationEventListenerBeanPostProcessor in your application context. This post processor will discover beans with @EventHandler annotated methods and automatically connect them to the event bus.

<beans xmlns="http://www.springframework.org/schema/beans">

    <bean class="org...AnnotationEventListenerBeanPostProcessor"> 
        <property name="eventBus" ref="eventBus"/> 
    </bean>

    <bean class="org.axonframework.sample.app.query.AddressTableUpdater"/> 

</beans>

You can also wire event listeners "manually", by explicitly defining them within a AnnotationEventListenerAdapter bean, as shown in the code sample below.

<beans xmlns="http://www.springframework.org/schema/beans">

    <bean class="org.axonframework...annotation.AnnotationEventListenerAdapter"> 
        <constructor-arg>
            <bean class="org.axonframework.sample.app.query.AddressTableUpdater"/>
        </constructor-arg>
        <property name="eventBus" ref="eventBus"/> 
    </bean>

</beans>

	Warning
	Be careful when wiring event listeners "manually" while there is also an AnnotationEventListenerBeanPostProcessor in the application context. This will cause the event listener to be wired twice.

Wiring command handlers is very much like wiring event handlers: there is an AnnotationCommandHandlerBeanPostProcessor which will automatically register classes containing command handler methods (i.e. methods annotated with the @CommandHandler annotation) with a command bus.

<beans xmlns="http://www.springframework.org/schema/beans">

    <bean class="org...AnnotationCommandHandlerBeanPostProcessor"> 
        <property name="commandBus" ref="commandBus"/> 
    </bean>

    <bean class="org.axonframework.sample.app.command.ContactCommandHandler"/> 

</beans>

As with event listeners, you can also wire command handlers "manually" by explicitly defining them within a AnnotationCommandHandlerAdapter bean, as shown in the code sample below.

<beans xmlns="http://www.springframework.org/schema/beans">

    <bean class="org.axonframework...annotation.AnnotationCommandHandlerAdapter"> 
        <constructor-arg>
            <bean class="org.axonframework.sample.app.command.ContactCommandHandler"/>
        </constructor-arg>
        <property name="commandBus" ref="commandBus"/> 
    </bean>

</beans>

	Warning
	Be careful when wiring command handlers "manually" while there is also an AnnotationCommandHandlerBeanPostProcessor in the application context. This will cause the command handler to be wired twice.

<bean class="org.axonframework.commandhandling.annotation.AggregateAnnotationCommandHandler"
      init-method="subscribe">
    <constructor-arg value="fully.qualified.AggregateClass"/>
    <constructor-arg ref="ref-to-repo"/>
    <constructor-arg ref="ref-to-command-bus"/>
</bean>

<!-- or, when using Namespace support -->

<axon:aggregate-command-handler aggregate-type="fully.qualified.AggregateClass" 
                                repository="ref-to-repo" 
                                command-bus="ref-to-command-bus"/>

The previous two sections explained how you wire bean post processors to activate annotation support for your command handlers and event listeners. Using support from the Axon namespace you can accomplish the same in one go, using the annotation-config element:

<axon:annotation-config />

The annotation-config element has the following attributes that allow you to configure annotation support further:

Using a Clustering Event Bus in Spring is very simple. In the Spring context, you just need to define the clusters you wish to use and tell them which Event Listeners you would like to be part of that Cluster. Axon will create the necessary infrastructure to assign listeners to the clusters.

<axon:cluster id="myFirstCluster">
    <axon:selectors>
        <axon:package prefix="com.mycompany.mypackage"/>
    </axon:selectors>
</axon:cluster>

<axon:cluster id="defaultCluster" default="true"/>

The example above will create two clusters. Event Listeners in the package com.mycompany.mypackage will be assigned to myFirstCluster, while all others are assigned to defaultCluster. Note that the latter does not have any selectors. Selectors are optional if the cluster is a default.

If there are conflicting selectors, and you would like to influence the order in which they are evaluated, you can add the order attribute to a cluster. Clusters with a lower value are evaluated before those with a higher value. Only if there are no matching selectors at all, Axon will assign a Listener to the Cluster with default="true". If no suitable cluster for any listener is found, Axon throws an exception.

	Organizing Cluster definitions in context files
	When you have an application that consists of a number of modules (represented in separate config files), it is possible to define the <axon:cluster> in the Context where the listeners are also defined. This makes the application more modular and less dependent on a centralized configuration.

To make a cluster replayable, simply add the <axon:replay-config> element to a <axon:cluster>, as in the example below:

<axon:cluster id="replayingCluster">
    <axon:replay-config event-store="eventStore" transaction-manager="mockTransactionManager"/>
    <axon:selectors>
        <axon:package prefix="com.mycompany.mypackage"/>
    </axon:selectors>
</axon:cluster>

The <axon:replay-config> element provides the necessary configuration to execute replays on a Cluster. The resulting Cluster bean will be of type ReplayingCluster.

It is also possible to use the <cluster> element while using a custom Cluster implementation:

<axon:cluster id="custoCluster">
    <bean class="com.mycompany.MyPersonalClusterImplementation"/>
    <axon:selectors>
        <axon:package prefix="com.mycompany.mypackage"/>
    </axon:selectors>
</axon:cluster>

If you have generated the tables automatically using your JPA implementation (e.g. Hibernate), you probably do not have all the right indexes set on your tables. Different usages of the Event Store require different indexes to be set for optimal performance. This list suggests the indexes that should be added for the different types of queries used by the default EventEntryStore implementation:

The default column lengths generated by e.g. Hibernate may work, but won't be optimal. A UUID, for example, will always have the same length. Instead of a variable length column of 255 characters, you could use a fixed length column of 36 characters for the aggregate identifier.

The 'timestamp' column in the DomainEventEntry table only stores ISO 8601 timestamps. If all times are stored in the UTZ timezone, they need a column length of 24 characters. If you use another timezone, this may be up to 28. Using variable length columns is generally not necessary, since time stamps always have the same length.

The 'type' column in the DomainEventEntry stores the Type Identifiers of aggregates. Generally, these are the 'simple name' of the aggregate. Event the infamous 'AbstractDependencyInjectionSpringContextTests' in spring only counts 45 characters. Here, again, a shorter (but variable) length field should suffice.

By default, the MongoEventStore will only generate the index it requires for correct operation. That means the required unique index on "Aggregate Identifier", "Aggregate Type" and "Event Sequence Number" is created when the Event Store is created. However, when using the MongoEventStore for certain opertaions, it might be worthwile to add some extra indices.

Note that there is always a balance between query optimization and update speed. Load testing is ultimately the best way to discover which indices provide the best performance.

Especially in distributed systems, Event Messages need to be serialized in multiple occasions. In the case of a Command Handling component that uses Event Sourcing, each message is serialized twice: once for the Event Store, and once to publish it on the Event Bus. Axon's components are aware of this and have support for SerializationAware messages. If a SerializationAware message is detected, its methods are used to serialize an object, instead of simply passing the payload to a serializer. This allows for performance optimizations.

By configuring the SerializationOptimizingInterceptor, all generated Events are wrapped into SerializationAware messages, and thus benefit from this optimization. Note that the optimization only helps if the same serializer is used for different components. If you use the DisruptorCommandBus, serialization can be optimized by providing a Serializer in the DisruptorConfiguration. The DisruptorCommandBus will then use an extra thread (or more when configured) to pre-serialize the Event Message using that serializer.

When you serialize messages yourself, and want to benefit from the SerializationAware optimization, use the MessageSerializer class to serialize the payload and meta data of messages. All optimization logic is implemented in that class. See the JavaDoc of the MessageSerializer for more details.