Package org.axonframework.disruptor.commandhandling

Class DisruptorCommandBus.Builder

java.lang.Object
org.axonframework.disruptor.commandhandling.DisruptorCommandBus.Builder
Enclosing class:
DisruptorCommandBus
public static class DisruptorCommandBus.Builder extends Object
Builder class to instantiate a DisruptorCommandBus.

The following configurable fields have defaults:

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.

  • Constructor Details

    • Builder

      public Builder()

  • Method Details

    • invokerInterceptors

      public DisruptorCommandBus.Builder invokerInterceptors(List<MessageHandlerInterceptor<? super CommandMessage<?>>> invokerInterceptors)
      Set the MessageHandlerInterceptor of generic type CommandMessage to use with the DisruptorCommandBus during in the invocation thread. The interceptors are invoked by the thread that also executes the command handler.

      Note that this is *not* the thread that stores and publishes the generated events. See publisherInterceptors(java.util.List).

      Parameters:
      invokerInterceptors - the MessageHandlerInterceptors to invoke when handling an incoming command
      Returns:
      the current Builder instance, for fluent interfacing

    • publisherInterceptors

      public DisruptorCommandBus.Builder publisherInterceptors(List<MessageHandlerInterceptor<CommandMessage<?>>> publisherInterceptors)
      Configures the MessageHandlerInterceptor of generic type CommandMessage to use with the DisruptorCommandBus during the publication of changes. The interceptors are invoked by the thread that also stores and publishes the events.
      Parameters:
      publisherInterceptors - the MessageHandlerInterceptors to invoke when handling an incoming command
      Returns:
      the current Builder instance, for fluent interfacing

    • dispatchInterceptors

      public DisruptorCommandBus.Builder dispatchInterceptors(List<MessageDispatchInterceptor<CommandMessage<?>>> dispatchInterceptors)
      Configures MessageDispatchInterceptor of generic type CommandMessage to use with the DisruptorCommandBus when commands are dispatched. The interceptors are invoked by the thread that provides the commands to the command bus.
      Parameters:
      dispatchInterceptors - the MessageDispatchInterceptors dispatch interceptors to invoke when dispatching a command
      Returns:
      the current Builder instance, for fluent interfacing

    • executor

      public DisruptorCommandBus.Builder executor(Executor executor)
      Sets the Executor that provides the processing resources (Threads) for the components of the DisruptorCommandBus. The provided executor must be capable of providing the required number of threads. Three threads are required immediately at startup and will not be returned until the CommandBus is stopped. Additional threads are used to invoke callbacks and start a recovery process in case aggregate state has been corrupted. Failure to do this results in the disruptor hanging at startup, waiting for resources to become available.

      Defaults to null, causing the DisruptorCommandBus to create the necessary threads itself. In that case, threads are created in the DisruptorCommandBus ThreadGroup.

      Parameters:
      executor - the Executor that provides the processing resources
      Returns:
      the current Builder instance, for fluent interfacing

    • rescheduleCommandsOnCorruptState

      public DisruptorCommandBus.Builder rescheduleCommandsOnCorruptState(boolean rescheduleCommandsOnCorruptState)
      Set the indicator specifying whether commands that failed because they were executed against potentially corrupted aggregate state should be automatically rescheduled. Commands that caused the aggregate state to become corrupted are never automatically rescheduled, to prevent poison message syndrome.

      Defaults to true.

      Parameters:
      rescheduleCommandsOnCorruptState - a boolean specifying whether or not to automatically reschedule commands that failed due to potentially corrupted aggregate state.
      Returns:
      the current Builder instance, for fluent interfacing

    • coolingDownPeriod

      public DisruptorCommandBus.Builder coolingDownPeriod(long coolingDownPeriod)
      Sets the cooling down period in milliseconds. This is the time in which new commands are no longer accepted, but the DisruptorCommandBus may reschedule commands that may have been executed against a corrupted Aggregate. If no commands have been rescheduled during this period, the disruptor shuts down completely. Otherwise, it wait until no commands were scheduled for processing.

      Defaults to 1000 ms (1 second).

      Parameters:
      coolingDownPeriod - a long specifying the cooling down period for the shutdown of the DisruptorCommandBus, in milliseconds.
      Returns:
      the current Builder instance, for fluent interfacing

    • commandTargetResolver

      public DisruptorCommandBus.Builder commandTargetResolver(CommandTargetResolver commandTargetResolver)
      Sets the CommandTargetResolver that must be used to indicate which Aggregate instance will be invoked by an incoming command. The DisruptorCommandBus only uses this value if invokerThreadCount(int)}, or publisherThreadCount(int) is greater than 1.

      Defaults to an AnnotationCommandTargetResolver instance.

      Parameters:
      commandTargetResolver - The CommandTargetResolver to use to indicate which Aggregate instance is target of an incoming Command
      Returns:
      the current Builder instance, for fluent interfacing

    • publisherThreadCount

      public DisruptorCommandBus.Builder publisherThreadCount(int publisherThreadCount)
      Sets the number of Threads that should be used to store and publish the generated Events. Defaults to 1.

      A good value for this setting mainly depends on the number of cores your machine has, as well as the amount of I/O that the process requires. If no I/O is involved, a good starting value is [processors / 2].

      Parameters:
      publisherThreadCount - the number of Threads to use for publishing as an int
      Returns:
      the current Builder instance, for fluent interfacing

    • messageMonitor

      public DisruptorCommandBus.Builder messageMonitor(MessageMonitor<? super CommandMessage<?>> messageMonitor)
      Sets the MessageMonitor of generic type CommandMessage used the to monitor the command bus. Defaults to a NoOpMessageMonitor.
      Parameters:
      messageMonitor - a MessageMonitor used the message monitor to monitor the command bus
      Returns:
      the current Builder instance, for fluent interfacing

    • transactionManager

      public DisruptorCommandBus.Builder transactionManager(TransactionManager transactionManager)
      Sets the TransactionManager to use to manage a transaction around the storage and publication of events. The default (null) is to not have publication and storage of events wrapped in a transaction.
      Parameters:
      transactionManager - the TransactionManager to use to manage a transaction around the storage and publication of events
      Returns:
      the current Builder instance, for fluent interfacing

    • rollbackConfiguration

      public DisruptorCommandBus.Builder rollbackConfiguration(RollbackConfiguration rollbackConfiguration)
      Sets the RollbackConfiguration which allows you to specify when a UnitOfWork should be rolled back. Defaults to a RollbackConfigurationType.UNCHECKED_EXCEPTIONS, which triggers a rollback on all unchecked exceptions.
      Parameters:
      rollbackConfiguration - a RollbackConfiguration specifying when a UnitOfWork should be rolled back
      Returns:
      the current Builder instance, for fluent interfacing

    • bufferSize

      public DisruptorCommandBus.Builder bufferSize(int bufferSize)
      Sets the buffer size to use. This field must be positive and a power of 2.

      The default is 4096.

      Parameters:
      bufferSize - an int specifying the buffer size to use
      Returns:
      the current Builder instance, for fluent interfacing

    • producerType

      public DisruptorCommandBus.Builder producerType(com.lmax.disruptor.dsl.ProducerType producerType)
      Sets the ProducerType to use by the Disruptor.

      Defaults to a ProducerType.MULTI solution.

      Parameters:
      producerType - the ProducerType to use by the Disruptor
      Returns:
      the current Builder instance, for fluent interfacing

    • waitStrategy

      public DisruptorCommandBus.Builder waitStrategy(com.lmax.disruptor.WaitStrategy waitStrategy)
      Sets the WaitStrategy which is used to make dependent threads wait for tasks to be completed. The choice of strategy mainly depends on the number of processors available and the number of tasks other than the DisruptorCommandBus being processed.

      The BusySpinWaitStrategy provides the best throughput at the lowest latency, but also put a big claim on available CPU resources. The SleepingWaitStrategy yields lower performance, but leaves resources available for other processes to use.

      Defaults to the BlockingWaitStrategy.

      Parameters:
      waitStrategy - The WaitStrategy to use
      Returns:
      the current Builder instance, for fluent interfacing
      See Also:
      • SleepingWaitStrategy
      • BlockingWaitStrategy
      • BusySpinWaitStrategy
      • YieldingWaitStrategy

    • invokerThreadCount

      public DisruptorCommandBus.Builder invokerThreadCount(int invokerThreadCount)
      Sets the number of Threads that should be used to invoke the Command Handlers. Defaults to 1.

      A good value for this setting mainly depends on the number of cores your machine has, as well as the amount of I/O that the process requires. A good range, if no I/O is involved is 1 .. ([processor count] / 2).

      Parameters:
      invokerThreadCount - an int specifying the number of Threads to use for Command Handler invocation
      Returns:
      the current Builder instance, for fluent interfacing

    • cache

      public DisruptorCommandBus.Builder cache(Cache cache)
      Sets the Cache in which loaded aggregates will be stored. Aggregates that are not active in the CommandBus' buffer will be loaded from this cache. If they are not in the cache, a new instance will be constructed using Events from the EventStore.

      By default, no cache is used.

      Parameters:
      cache - the cache to store loaded aggregates in
      Returns:
      the current Builder instance, for fluent interfacing

    • duplicateCommandHandlerResolver

      public DisruptorCommandBus.Builder duplicateCommandHandlerResolver(DuplicateCommandHandlerResolver duplicateCommandHandlerResolver)
      Sets the DuplicateCommandHandlerResolver used to resolves the road to take when a duplicate command handler is subscribed. Defaults to Log and Override.
      Parameters:
      duplicateCommandHandlerResolver - a DuplicateCommandHandlerResolver used to resolves the road to take when a duplicate command handler is subscribed
      Returns:
      the current Builder instance, for fluent interfacing

    • defaultCommandCallback

      public DisruptorCommandBus.Builder defaultCommandCallback(CommandCallback<Object,Object> defaultCommandCallback)
      Sets the callback to use when commands are dispatched in a "fire and forget" method, such as DisruptorCommandBus.dispatch(CommandMessage). Defaults to a DisruptorCommandBus.FailureLoggingCommandCallback, which logs failed commands to a logger. Passing null will result in a NoOpCallback being used.
      Parameters:
      defaultCommandCallback - the callback to invoke when no explicit callback is provided for a command
      Returns:
      the current Builder instance, for fluent interfacing

    • build

      public DisruptorCommandBus build()
      Initializes a DisruptorCommandBus as specified through this Builder.
      Returns:
      a DisruptorCommandBus as specified through this Builder

    • validate

      protected void validate()
      Validate whether the fields contained in this Builder as set accordingly.
      Throws:
      AxonConfigurationException - if one field is asserted to be incorrect according to the Builder's specifications