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.

At this moment, there is a single implementation of the Cluster interface: SimpleCluster. This implementation calls each EventListener sequentially in the calling thread. By adding information in the Meta Data of a cluster, the selector can provide hints to the Terminal about the expected behavior.

The EventBusTerminal forms the bridge between the events being dispatched and the Clusters inside the Event Bus. The terminal is aware of any remoting technologies used, such as JMS, AMQP, etc. The default implementation dispatches published events to each of the (local) clusters using the publishing thread. This means that with the default terminal, and the default ClusterSelector, the behavior of the ClusteringEventBus is exactly the same as that of the SimpleEventBus.

In a typical AMQP based configuration, the EventBusTerminal would send published events to an Exchange. For each cluster, a Queue would be connected to that exchange. The EventBusTerminal will create a consumer for each cluster, which reads from its related Queue and forwards each message to that cluster. Event Listeners in a distributed environment where at most one instance should receive an Events should be placed in a separate cluster, which competes with the other instances on a single Queue.

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 and are resolved according to the same rules that count for annotated aggregate roots (see the section called “ AbstractAnnotatedAggregateRoot ”).

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.

	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 inside an AnnotationEventListenerAdapter and register them with an event bus.

If you use the AnnotationEventListenerAdapter, or <axon:annotation-config/>, an Executor must be configured to allow asynchronous processing of events.

You can configure the event sequencing policy on the @AsynchronousEventListener annotation. You then set the sequencePolicyClass to the type of policy you like to use. Note that you can only choose policy classes that provide a public no-arg constructor.

@AsynchronousEventListener(sequencingPolicyClass = MyCustomPolicy.class)
public class MyEventListener() {

    @EventHandler
    public void onSomeImportantEvent(MyEvent event) {
        // eventProcessing logic
    }
}

public class MyCustomPolicy implements EventSequencingPolicy {
    public Object getSequenceIdentifierFor(Event event) {
        if (event instanceof MyEvent) {
            // let's assume that we do processing based on the someProperty field.
            return ((MyEvent) event).someProperty();
        }
        return null;
    }
}                    

With annotation support, the event handler bean must also act as a transaction manager in order to support transactions. There is annotation support for transaction management, too (see Section 6.2.3, “Managing transactions in asynchronous event handling”).

There are a number of settings you can use on the TransactionStatus object.

You can configure a yielding policy, which gives the scheduler an indication of that to do when a batch has finished, but more events are available for processing. Use DO_NOT_YIELD if you want the scheduler to continue processing immediately as long as new events are available for processing. The YIELD_AFTER_TRANSACTION policy will tell the scheduler to reschedule the next batch for processing when a thread is available. The first will make sure events are processed earlier, while the latter provides a fairer execution of events, as yielding provides waiting thread a chance to start processing. The choice of yielding policy should be driven by the SLA of the event listener.

You can set the maximum number of events to handle within a transaction using setMaxTransactionSize(int). The default of this value is the number of events ready for processing at the moment the transaction started.

When an event handler throws an exception, for example because a data source is not available, the transaction is marked as failed. In that case, isSuccessful() on the TransactionStatus object will return false and getException() will return the exception that the scheduler caught. It is the responsibility of the event listener to rollback or commit any active underlying transactions, based on the information provided by these methods.

The event handler can provide a policy setRetryPolicy(RetryPolicy) to tell the scheduler what to do in such case. There are three policies, each for a specific scenario:

Note that the SKIP_FAILED_EVENT is the default policy. For event handlers that use an underlying mechanism to perform actions, this might not be a suitable policy. Exceptions resulting from errors in these underlying systems (such as databases or email clients) would cause events to be ignored when the underlying system is unavailable. In error situations, the event listener should inspect the exception (using the getException() method) and decide whether it makes sense to retry processing of this event. If that is the case, it should set the RETRY_LAST_EVENT or RETRY_TRANSACTION policy, depending on the transactional behavior of the underlying system.

When the chosen policy forces a retry of event processing, the processing is delayed by the number of milliseconds defined in the retryInterval property. The default interval is 5 seconds.

You can change transaction semantics event during event processing. This can be done in one of two ways, depending on the type of event handler you use.

If you use the @EventHandler annotation to mark event handler methods, you may use a second parameter of type TransactionStatus. If such parameter is available on the annotated method, the current TransactionStatus object is passed as a parameter.

Alternatively, you can use the static TransactionStatus.current() accessor to gain access to the status of the current transaction. Note that this method returns null if there is no active transaction.

With the current transaction status, you can use the requestImmediateYield() and requestImmediateCommit() methods to end the transaction after processing of the event. The former will also tell the scheduler to reschedule the remainder of the events for another batch. The latter will use the yield policy to see what needs to be done. Since the default yielding policy is YIELD_AFTER_TRANSACTION, the behavior of both methods is identical when using these defaults.

As with many of the other supported features in Axon, there is also annotation support for transaction management. You have several options to configure transactions.

The first is to annotate methods on your EventListener with @BeforeTransaction and @AfterTransaction. These methods will be called before and after the execution of a transactional batch, respectively. The annotated methods may accept a single parameter of type TransactionStatus, which provides access to transaction details, such as current status and configuration.

Alternatively, you can use an external Transaction Manager, which you assign to a field. If you annotate that field with @TransactionManager, Axon will autodetect it and use it as transaction manager for that listener. The transaction manager may be either one that implements the TransactionManager interface, or any other type that uses annotations.

Currently, Axon Framework provides one TransactionManager implementation, the SpringTransactionManager. This implemenation uses Spring's PlatformTransactionManager as underlying transaction mechanism. That means the SpringTransactionManager can manage any transactions in resources that Spring supports.