org.axonframework.commandhandling.distributed

Class DistributedCommandBus

java.lang.Object

org.axonframework.commandhandling.distributed.DistributedCommandBus

All Implemented Interfaces:

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

public class DistributedCommandBus
extends Object
implements CommandBus, Distributed<CommandBus>, Lifecycle

Implementation of a CommandBus that is aware of multiple instances of a CommandBus working together to spread load. Each "physical" CommandBus instance is considered a "segment" of a conceptual distributed CommandBus.

The DistributedCommandBus relies on a CommandBusConnector to dispatch commands and replies to different segments of the CommandBus. Depending on the implementation used, each segment may run in a different JVM.

Since:

2.0

Author:

Allard Buijze

Nested Class Summary

Nested Classes

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

Nested classes/interfaces inherited from interface org.axonframework.lifecycle.Lifecycle

Lifecycle.LifecycleHandler, Lifecycle.LifecycleRegistry

Field Summary

Fields

Modifier and Type	Field and Description
static int	INITIAL_LOAD_FACTOR The initial load factor of this node when it is registered with the CommandRouter.

Constructor Summary

Constructors

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

Method Summary

Modifier and Type	Method and Description
static DistributedCommandBus.Builder	builder() Instantiate a Builder to be able to create a DistributedCommandBus.
void	disconnect() Disconnect the command bus for receiving new commands, by unsubscribing all registered command handlers.
<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.
int	getLoadFactor() Returns the current load factor of this node.
CommandBus	localSegment() Return the message bus of type MessageBus which is regarded as the local segment for this implementation.
Registration	registerDispatchInterceptor(MessageDispatchInterceptor<? super CommandMessage<?>> dispatchInterceptor) Registers the given list of dispatch interceptors to the command bus.
Registration	registerHandlerInterceptor(MessageHandlerInterceptor<? super CommandMessage<?>> handlerInterceptor) Register the given handlerInterceptor.
void	registerLifecycleHandlers(Lifecycle.LifecycleRegistry handle) Registers the activities to be executed in the various phases of an application's lifecycle.
CompletableFuture<Void>	shutdownDispatching() Shutdown the command bus asynchronously for dispatching commands to other instances.
Registration	subscribe(String commandName, MessageHandler<? super CommandMessage<?>> handler) Subscribe the given handler to commands with the given commandName.
void	updateLoadFactor(int loadFactor) Updates the load factor of this node compared to other nodes registered with the CommandRouter.

Methods inherited from class java.lang.Object

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

Field Detail

INITIAL_LOAD_FACTOR

public static final int INITIAL_LOAD_FACTOR

The initial load factor of this node when it is registered with the CommandRouter.

See Also:

Constant Field Values

Constructor Detail

DistributedCommandBus

protected DistributedCommandBus(DistributedCommandBus.Builder builder)

Instantiate a DistributedCommandBus based on the fields contained in the DistributedCommandBus.Builder.

Will assert that the CommandRouter, CommandBusConnector and MessageMonitor are not null, and will throw an AxonConfigurationException if any of them is null.

Parameters:

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

Method Detail

builder

public static DistributedCommandBus.Builder builder()

Instantiate a Builder to be able to create a DistributedCommandBus.

The CommandCallback is defaulted to a LoggingCallback. The MessageMonitor is defaulted to a NoOpMessageMonitor. The CommandBusSpanFactory is defaulted to a DefaultCommandBusSpanFactory backed by a NoOpSpanFactory. The CommandRouter and CommandBusConnector are hard requirements and as such should be provided.

Returns:

a Builder to be able to create a DistributedCommandBus

disconnect

public void disconnect()

Disconnect the command bus for receiving new commands, by unsubscribing all registered command handlers. This shutdown operation is performed in the Phase.INBOUND_COMMAND_CONNECTOR phase.

shutdownDispatching

public CompletableFuture<Void> shutdownDispatching()

Shutdown the command bus asynchronously for dispatching commands to other instances. This process will wait for dispatched commands which have not received a response yet. This shutdown operation is performed in the Phase.OUTBOUND_COMMAND_CONNECTORS phase.

Returns:

a completable future which is resolved once all command dispatching activities are completed

registerLifecycleHandlers

public void registerLifecycleHandlers(@Nonnull
                                      Lifecycle.LifecycleRegistry handle)

Description copied from interface: Lifecycle

Registers the activities to be executed in the various phases of an application's lifecycle. This could either be at startup, shutdown, or both.

Specified by:

registerLifecycleHandlers in interface Lifecycle

Parameters:

handle - the lifecycle instance to register the handlers with

See Also:

Lifecycle.LifecycleRegistry.onShutdown(int, Runnable), LifecycleRegistry#onShutdown(int, LifecycleHandler), Lifecycle.LifecycleRegistry.onStart(int, Runnable), LifecycleRegistry#onStart(int, LifecycleHandler)

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)

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

Throws:

CommandDispatchException - when an error occurs while dispatching the command to a segment

See Also:

GenericCommandMessage.asCommandMessage(Object)

subscribe

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

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.

In the DistributedCommandBus, the handler is subscribed to the local segment only.

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.

localSegment

public CommandBus localSegment()

Return the message bus of type MessageBus which is regarded as the local segment for this implementation. Would return the message bus used to dispatch and handle messages in a local environment to bridge the gap in a distributed set up.

Will call CommandBusConnector.localSegment(). If this returns an Optional.empty(), this method defaults to returning this as last resort.

Specified by:

localSegment in interface Distributed<CommandBus>

Returns:

a MessageBus which is the local segment for this distributed message bus implementation

getLoadFactor

public int getLoadFactor()

Returns the current load factor of this node.

Returns:

the current load factor

updateLoadFactor

public void updateLoadFactor(int loadFactor)

Updates the load factor of this node compared to other nodes registered with the CommandRouter.

Parameters:

loadFactor - the new load factor of this node

registerDispatchInterceptor

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

Registers the given list of dispatch interceptors to the command bus. All incoming commands will pass through the interceptors at the given order before the command is dispatched toward the command handler.

Specified by:

registerDispatchInterceptor in interface MessageDispatchInterceptorSupport<CommandMessage<?>>

Parameters:

dispatchInterceptor - The interceptors to invoke when commands are dispatched

Returns:

handle 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.