wow-core/me.ahoo.wow.modeling.command.dispatcher/AggregateCommandDispatcher

AggregateCommandDispatcher

class AggregateCommandDispatcher<C : Any, S : Any>(val name: String = "-", val aggregateMetadata: AggregateMetadata<C, S>, val messageFlux: Flux<ServerCommandExchange<*>>, val parallelism: Int = MessageParallelism.DEFAULT_PARALLELISM, commandHandler: CommandHandler, val scheduler: Scheduler, messageReadiness: Mono<Void> = Mono.empty(), processingAdmission: () -> Unit = {}, processingQuiescence: () -> Unit = {}, metrics: WowMetrics = WowMetrics.NONE) : AggregateDispatcher<ServerCommandExchange<*>>

Aggregate command dispatcher grouped by named aggregate.

This dispatcher manages command processing for a specific named aggregate, ensuring proper parallelism and thread affinity. Each aggregate ID is bound to one worker thread, but one worker can handle multiple aggregate IDs, providing efficient resource utilization.

Key characteristics:

One AggregateId binds to one Worker (Thread)
One Worker can be bound by multiple aggregateIds
Workers have aggregate ID affinity for consistent processing

Parameters

name

The name of this dispatcher.

messageFlux

The flux of command exchanges to process.

parallelism

The level of parallelism for message processing.

commandHandler

The command handler for processing commands.

scheduler

The scheduler for handling messages.

messageReadiness

Completion of asynchronous message-source setup when this dispatcher is registered directly with a runtime.

processingAdmission

Explicit transport-processing gate opened by start.

processingQuiescence

Logical transport gate closed by quiesce.

metrics

Instance-scoped metrics recorder for dispatcher operations.

Type Parameters

The type of the command aggregate root.

The type of the state aggregate.

Constructors

AggregateCommandDispatcher

constructor(name: String = "-", aggregateMetadata: AggregateMetadata<C, S>, messageFlux: Flux<ServerCommandExchange<*>>, parallelism: Int = MessageParallelism.DEFAULT_PARALLELISM, commandHandler: CommandHandler, scheduler: Scheduler, messageReadiness: Mono<Void> = Mono.empty(), processingAdmission: () -> Unit = {}, processingQuiescence: () -> Unit = {}, metrics: WowMetrics = WowMetrics.NONE)