Package org.axonframework.disruptor.commandhandling

Class DisruptorCommandBus

java.lang.Object

org.axonframework.disruptor.commandhandling.DisruptorCommandBus

All Implemented Interfaces:

CommandBus, MessageDispatchInterceptorSupport<CommandMessage<?>>, MessageHandlerInterceptorSupport<CommandMessage<?>>

public class DisruptorCommandBus
extends Object
implements CommandBus

Asynchronous CommandBus implementation with very high performance characteristics. It divides the command handling process in two steps, which can be executed in different threads. The CommandBus is backed by a Disruptor, which ensures that two steps are executed sequentially in these threads, while minimizing locking and inter-thread communication.

The process is split into two separate steps, each of which is executed in a different thread:

1. Command Handler execution
   This process invokes the command handler with the incoming command. The result and changes to the aggregate are recorded for the next step.
2. Event storage and publication
   This process stores all generated domain events and publishes them (with any optional application events) to the event bus. Finally, an asynchronous task is scheduled to invoke the command handler callback with the result of the command handling result.

Exceptions and recovery

This separation of process steps makes this implementation very efficient and highly performing. However, it does not cope with exceptions very well. When an exception occurs, an Aggregate that has been loaded is potentially corrupt. That means that an aggregate does not represent a state that can be reproduced by replaying its committed events. Although this implementation will recover from this corrupt state, it may result in a number of commands being rejected in the meantime. These command may be retried if the cause of the AggregateStateCorruptedException does not indicate a non-transient error.

Commands that have been executed against a potentially corrupt Aggregate will result in a AggregateStateCorruptedException exception. These commands are automatically rescheduled for processing by default. Use DisruptorCommandBus.Builder.rescheduleCommandsOnCorruptState(boolean) to disable this feature. Note that the order in which commands are executed is not fully guaranteed when this feature is enabled (default). Limitations of this implementation

Although this implementation allows applications to achieve extreme performance (over 1M commands on commodity hardware), it does have some limitations. It only allows a single aggregate to be invoked during command processing.

This implementation can only work with Event Sourced Aggregates. Infrastructure considerations

This CommandBus implementation has special requirements for the Repositories being used during Command Processing. Therefore, the Repository instance to use in the Command Handler must be created using createRepository(EventStore, AggregateFactory, RepositoryProvider). Using another repository will most likely result in undefined behavior.

The DisruptorCommandBus must have access to at least 3 threads, two of which are permanently used while the DisruptorCommandBus is operational. At least one additional thread is required to invoke callbacks and initiate a recovery process in the case of exceptions.

Consider providing an alternative IdentifierFactory implementation. The default implementation used UUID.randomUUID() to generated identifier for Events. The poor performance of this method severely impacts overall performance of the DisruptorCommandBus. A better performing alternative is, for example, com.eaio.uuid.UUID

Since:

2.0

Author:

Allard Buijze

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	DisruptorCommandBus.Builder	Builder class to instantiate a DisruptorCommandBus.

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	DisruptorCommandBus(DisruptorCommandBus.Builder builder)	Instantiate a DisruptorCommandBus based on the fields contained in the DisruptorCommandBus.Builder.

Method Summary

Modifier and Type	Method	Description
static DisruptorCommandBus.Builder	builder()	Instantiate a Builder to be able to create a DisruptorCommandBus.
<T> Repository<T>	createRepository(EventStore eventStore, AggregateFactory<T> aggregateFactory)	Creates a repository instance for an Event Sourced aggregate that is created by the given eventStore and aggregateFactory.
<T> Repository<T>	createRepository(EventStore eventStore, AggregateFactory<T> aggregateFactory, SnapshotTriggerDefinition snapshotTriggerDefinition)	Creates a repository instance for an Event Sourced aggregate, source from given eventStore, that is created by the given aggregateFactory.
<T> Repository<T>	createRepository(EventStore eventStore, AggregateFactory<T> aggregateFactory, SnapshotTriggerDefinition snapshotTriggerDefinition, ParameterResolverFactory parameterResolverFactory)	Creates a repository instance for an Event Sourced aggregate, sourced from given eventStore, that is created by the given aggregateFactory.
<T> Repository<T>	createRepository(EventStore eventStore, AggregateFactory<T> aggregateFactory, SnapshotTriggerDefinition snapshotTriggerDefinition, ParameterResolverFactory parameterResolverFactory, HandlerDefinition handlerDefinition, RepositoryProvider repositoryProvider)	Creates a repository instance for an Event Sourced aggregate, sourced from given eventStore, that is created by the given aggregateFactory.
<T> Repository<T>	createRepository(EventStore eventStore, AggregateFactory<T> aggregateFactory, SnapshotTriggerDefinition snapshotTriggerDefinition, RepositoryProvider repositoryProvider)	Creates a repository instance for an Event Sourced aggregate, source from given eventStore, that is created by the given aggregateFactory.
<T> Repository<T>	createRepository(EventStore eventStore, AggregateFactory<T> aggregateFactory, ParameterResolverFactory parameterResolverFactory)	Creates a repository instance for an Event Sourced aggregate that is created by the given aggregateFactory and sourced from given eventStore.
<T> Repository<T>	createRepository(EventStore eventStore, AggregateFactory<T> aggregateFactory, ParameterResolverFactory parameterResolverFactory, HandlerDefinition handlerDefinition, RepositoryProvider repositoryProvider)	Creates a repository instance for an Event Sourced aggregate that is created by the given aggregateFactory and sourced from given eventStore.
<T> Repository<T>	createRepository(EventStore eventStore, AggregateFactory<T> aggregateFactory, RepositoryProvider repositoryProvider)	Creates a repository instance for an Event Sourced aggregate that is created by the given eventStore and aggregateFactory.
<C> void	dispatch(CommandMessage<C> command)	Dispatch the given command to the CommandHandler subscribed to the given command's name.
<C, R> void	dispatch(CommandMessage<C> command, CommandCallback<? super C,? super R> callback)	Dispatch the given command to the CommandHandler subscribed to the given command's name.
Registration	registerDispatchInterceptor(MessageDispatchInterceptor<? super CommandMessage<?>> dispatchInterceptor)	Register the given DispatchInterceptor.
Registration	registerHandlerInterceptor(MessageHandlerInterceptor<? super CommandMessage<?>> handlerInterceptor)	Register the given handlerInterceptor.
void	stop()	Shuts down the command bus.
Registration	subscribe(String commandName, MessageHandler<? super CommandMessage<?>> handler)	Subscribe the given handler to commands with the given commandName.

Methods inherited from class java.lang.Object

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

Constructor Details

DisruptorCommandBus

protected DisruptorCommandBus(DisruptorCommandBus.Builder builder)

Instantiate a DisruptorCommandBus based on the fields contained in the DisruptorCommandBus.Builder. The Threads required for command execution are immediately requested from the Configuration's Executor, if any. Otherwise, they are created.

Will assert that the CommandTargetResolver, MessageMonitor, RollbackConfiguration, ProducerType, WaitStrategy and Cache are not null. Additional verification is done on the the coolingDownPeriod, publisherThreadCount, bufferSize and invokerThreadCount to check whether they are positive numbers. If any of these checks fails, an AxonConfigurationException will be thrown.

Parameters:

builder - the DisruptorCommandBus.Builder used to instantiate a DisruptorCommandBus instance

Method Details

builder

public static DisruptorCommandBus.Builder builder()

Instantiate a Builder to be able to create a DisruptorCommandBus.

The following configurable fields have defaults:

• The rescheduleCommandsOnCorruptState defaults to true.
• The coolingDownPeriod defaults to 1000.
• The CommandTargetResolver defaults to an AnnotationCommandTargetResolver.
• The publisherThreadCount defaults to 1.
• The MessageMonitor defaults to NoOpMessageMonitor.INSTANCE.
• The RollbackConfiguration defaults to RollbackConfigurationType.UNCHECKED_EXCEPTIONS.
• The bufferSize defaults to 4096.
• The ProducerType defaults to ProducerType.MULTI.
• The WaitStrategy defaults to a BlockingWaitStrategy.
• The invokerThreadCount defaults to 1.
• The Cache defaults to NoCache.INSTANCE.
• The DuplicateCommandHandlerResolver defaults to DuplicateCommandHandlerResolution.logAndOverride().

The (2) Threads required for command execution are created immediately. Additional threads are used to invoke response callbacks and to initialize a recovery process in the case of errors. The thread creation process can be specified by providing an Executor.

The CommandTargetResolver, MessageMonitor, RollbackConfiguration, ProducerType, WaitStrategy and Cache are a hard requirements. Thus setting them to null will result in an AxonConfigurationException. Additionally, the coolingDownPeriod, publisherThreadCount, bufferSize and invokerThreadCount have a positive number constraint, thus will also result in an AxonConfigurationException if set otherwise.

Returns:

a Builder to be able to create a DisruptorCommandBus

dispatch

public <C> void dispatch(@Nonnull
 CommandMessage<C> command)

Description copied from interface: CommandBus

Dispatch the given command to the CommandHandler subscribed to the given command's name. No feedback is given about the status of the dispatching process. Implementations may return immediately after asserting a valid handler is registered for the given command.

Specified by:

dispatch in interface CommandBus

Type Parameters:

C - The payload type of the command to dispatch

Parameters:

command - The Command to dispatch

See Also:

• GenericCommandMessage.asCommandMessage(Object)

dispatch

public <C,
R> void dispatch(@Nonnull
 CommandMessage<C> command,
 @Nonnull
 CommandCallback<? super C,? super R> callback)

Description copied from interface: CommandBus

Dispatch the given command to the CommandHandler subscribed to the given command's name. When the command is processed, one of the callback's methods is called, depending on the result of the processing.

There are no guarantees about the successful completion of command dispatching or handling after the method returns. Implementations are highly recommended to perform basic validation of the command before returning from this method call.

Implementations must start a UnitOfWork when before dispatching the command, and either commit or rollback after a successful or failed execution, respectively.

Specified by:

dispatch in interface CommandBus

Type Parameters:

C - The payload type of the command to dispatch

R - The type of the expected result

Parameters:

command - The Command to dispatch

callback - The callback to invoke when command processing is complete

See Also:

• GenericCommandMessage.asCommandMessage(Object)

createRepository

public <T> Repository<T> createRepository(EventStore eventStore,
 AggregateFactory<T> aggregateFactory)

Creates a repository instance for an Event Sourced aggregate that is created by the given eventStore and aggregateFactory.

The repository returned must be used by Command Handlers subscribed to this Command Bus for loading aggregate instances. Using any other repository instance may result in undefined outcome (a.k.a. concurrency problems).

Type Parameters:

T - The type of aggregate to create the repository for

Parameters:

eventStore - The Event Store to retrieve and persist events

aggregateFactory - The factory creating uninitialized instances of the Aggregate

Returns:

the repository that provides access to stored aggregates

createRepository

public <T> Repository<T> createRepository(EventStore eventStore,
 AggregateFactory<T> aggregateFactory,
 RepositoryProvider repositoryProvider)

Creates a repository instance for an Event Sourced aggregate that is created by the given eventStore and aggregateFactory.

The repository returned must be used by Command Handlers subscribed to this Command Bus for loading aggregate instances. Using any other repository instance may result in undefined outcome (a.k.a. concurrency problems).

Type Parameters:

T - The type of aggregate to create the repository for

Parameters:

eventStore - The Event Store to retrieve and persist events

aggregateFactory - The factory creating uninitialized instances of the Aggregate

repositoryProvider - Provides repositories for specified aggregate types

Returns:

the repository that provides access to stored aggregates

createRepository

public <T> Repository<T> createRepository(EventStore eventStore,
 AggregateFactory<T> aggregateFactory,
 SnapshotTriggerDefinition snapshotTriggerDefinition)

Creates a repository instance for an Event Sourced aggregate, source from given eventStore, that is created by the given aggregateFactory.

The repository returned must be used by Command Handlers subscribed to this Command Bus for loading aggregate instances. Using any other repository instance may result in undefined outcome (a.k.a. concurrency problems).

Type Parameters:

T - The type of aggregate to create the repository for

Parameters:

eventStore - The Event Store to retrieve and persist events

aggregateFactory - The factory creating uninitialized instances of the Aggregate

snapshotTriggerDefinition - The trigger definition for creating snapshots

Returns:

the repository that provides access to stored aggregates

createRepository

public <T> Repository<T> createRepository(EventStore eventStore,
 AggregateFactory<T> aggregateFactory,
 SnapshotTriggerDefinition snapshotTriggerDefinition,
 RepositoryProvider repositoryProvider)

Creates a repository instance for an Event Sourced aggregate, source from given eventStore, that is created by the given aggregateFactory. The given decorator is used to decorate event streams.

The repository returned must be used by Command Handlers subscribed to this Command Bus for loading aggregate instances. Using any other repository instance may result in undefined outcome (a.k.a. concurrency problems).

Note that a second invocation of this method with an aggregate factory for the same aggregate type may return the same instance as the first invocation, even if the given decorator is different.

Type Parameters:

T - The type of aggregate to create the repository for

Parameters:

eventStore - The Event Store to retrieve and persist events

aggregateFactory - The factory creating uninitialized instances of the Aggregate

snapshotTriggerDefinition - The trigger definition for creating snapshots

repositoryProvider - Provides repositories for specified aggregate types

Returns:

the repository that provides access to stored aggregates

createRepository

public <T> Repository<T> createRepository(EventStore eventStore,
 AggregateFactory<T> aggregateFactory,
 ParameterResolverFactory parameterResolverFactory)

Creates a repository instance for an Event Sourced aggregate that is created by the given aggregateFactory and sourced from given eventStore. Parameters of the annotated methods are resolved using the given parameterResolverFactory.

Type Parameters:

T - The type of aggregate managed by this repository

Parameters:

eventStore - The Event Store to retrieve and persist events

aggregateFactory - The factory creating uninitialized instances of the Aggregate

parameterResolverFactory - The ParameterResolverFactory to resolve parameter values of annotated handler with

Returns:

the repository that provides access to stored aggregates

createRepository

public <T> Repository<T> createRepository(EventStore eventStore,
 AggregateFactory<T> aggregateFactory,
 ParameterResolverFactory parameterResolverFactory,
 HandlerDefinition handlerDefinition,
 RepositoryProvider repositoryProvider)

Creates a repository instance for an Event Sourced aggregate that is created by the given aggregateFactory and sourced from given eventStore. Parameters of the annotated methods are resolved using the given parameterResolverFactory. The given handlerDefinition is used to create handler instances.

Type Parameters:

T - The type of aggregate managed by this repository

Parameters:

eventStore - The Event Store to retrieve and persist events

aggregateFactory - The factory creating uninitialized instances of the Aggregate

parameterResolverFactory - The ParameterResolverFactory to resolve parameter values of annotated handler with

handlerDefinition - The handler definition used to create concrete handlers

repositoryProvider - Provides specific for given aggregate types

Returns:

the repository that provides access to stored aggregates

createRepository

public <T> Repository<T> createRepository(EventStore eventStore,
 AggregateFactory<T> aggregateFactory,
 SnapshotTriggerDefinition snapshotTriggerDefinition,
 ParameterResolverFactory parameterResolverFactory)

Creates a repository instance for an Event Sourced aggregate, sourced from given eventStore, that is created by the given aggregateFactory. Parameters of the annotated methods are resolved using the given parameterResolverFactory.

Type Parameters:

T - The type of aggregate managed by this repository

Parameters:

eventStore - The Event Store to retrieve and persist events

aggregateFactory - The factory creating uninitialized instances of the Aggregate

snapshotTriggerDefinition - The trigger definition for snapshots

parameterResolverFactory - The ParameterResolverFactory to resolve parameter values of annotated handler with

Returns:

the repository that provides access to stored aggregates

createRepository

public <T> Repository<T> createRepository(EventStore eventStore,
 AggregateFactory<T> aggregateFactory,
 SnapshotTriggerDefinition snapshotTriggerDefinition,
 ParameterResolverFactory parameterResolverFactory,
 HandlerDefinition handlerDefinition,
 RepositoryProvider repositoryProvider)

Creates a repository instance for an Event Sourced aggregate, sourced from given eventStore, that is created by the given aggregateFactory. Parameters of the annotated methods are resolved using the given parameterResolverFactory. The given handlerDefinition is used to create handler instances.

Type Parameters:

T - The type of aggregate managed by this repository

Parameters:

eventStore - The Event Store to retrieve and persist events

aggregateFactory - The factory creating uninitialized instances of the Aggregate

snapshotTriggerDefinition - The trigger definition for snapshots

parameterResolverFactory - The ParameterResolverFactory to resolve parameter values of annotated handler with

handlerDefinition - The handler definition used to create concrete handlers

repositoryProvider - Provides repositories for specific aggregate types

Returns:

the repository that provides access to stored aggregates

subscribe

public Registration subscribe(@Nonnull
 String commandName,
 @Nonnull
 MessageHandler<? super CommandMessage<?>> handler)

Description copied from interface: CommandBus

Subscribe the given handler to commands with the given commandName.

If a subscription already exists for the given name, the behavior is undefined. Implementations may throw an Exception to refuse duplicate subscription or alternatively decide whether the existing or new handler gets the subscription.

Specified by:

subscribe in interface CommandBus

Parameters:

commandName - The name of the command to subscribe the handler to

handler - The handler instance that handles the given type of command

Returns:

a handle to unsubscribe the handler. When unsubscribed it will no longer receive commands.

stop

public void stop()

Shuts down the command bus. It no longer accepts new commands, and finishes processing commands that have already been published. This method will not shut down any executor that has been provided as part of the Builder process.

registerDispatchInterceptor

@Nonnull
public Registration registerDispatchInterceptor(@Nonnull
 MessageDispatchInterceptor<? super CommandMessage<?>> dispatchInterceptor)

Description copied from interface: MessageDispatchInterceptorSupport

Register the given DispatchInterceptor. After registration, the interceptor will be invoked for each Message dispatched on the messaging component that it was registered to.

Specified by:

registerDispatchInterceptor in interface MessageDispatchInterceptorSupport<CommandMessage<?>>

Parameters:

dispatchInterceptor - The interceptor to register

Returns:

A Registration, which may be used to deregister the interceptor.

registerHandlerInterceptor

public Registration registerHandlerInterceptor(@Nonnull
 MessageHandlerInterceptor<? super CommandMessage<?>> handlerInterceptor)

Description copied from interface: MessageHandlerInterceptorSupport

Register the given handlerInterceptor. After registration, the interceptor will be invoked for each handled Message on the messaging component that it was registered to, prior to invoking the message's handler.

Specified by:

registerHandlerInterceptor in interface MessageHandlerInterceptorSupport<CommandMessage<?>>

Parameters:

handlerInterceptor - The interceptor to register

Returns:

A Registration, which may be used to deregister the interceptor.