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

AggregateCommandDispatcher

class AggregateCommandDispatcher<C : Any, S : Any>(val aggregateMetadata: AggregateMetadata<C, S>, val parallelism: Int = MessageParallelism.DEFAULT_PARALLELISM, val scheduler: Scheduler, val messageFlux: Flux<ServerCommandExchange<*>>, val name: String = "-", aggregateProcessorFactory: AggregateProcessorFactory, commandHandler: CommandHandler, serviceProvider: ServiceProvider) : AggregateMessageDispatcher<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

C

The type of the command aggregate root.

S

The type of the state aggregate.

parallelism

The level of parallelism for message processing.

scheduler

The scheduler for handling messages.

messageFlux

The flux of command exchanges to process.

name

The name of this dispatcher.

aggregateProcessorFactory

Factory for creating aggregate processors.

commandHandler

The command handler for processing commands.

serviceProvider

Provider for accessing services.

Constructors

AggregateCommandDispatcher

constructor(aggregateMetadata: AggregateMetadata<C, S>, parallelism: Int = MessageParallelism.DEFAULT_PARALLELISM, scheduler: Scheduler, messageFlux: Flux<ServerCommandExchange<*>>, name: String = "-", aggregateProcessorFactory: AggregateProcessorFactory, commandHandler: CommandHandler, serviceProvider: ServiceProvider)

Properties

aggregateMetadata

val aggregateMetadata: AggregateMetadata<C, S>

The metadata for the aggregate being dispatched.

open override val aggregateName: String

open override val contextName: String

open override val isDisposed: Boolean

open override val messageFlux: Flux<ServerCommandExchange<*>>

open override val name: String

open override val namedAggregate: NamedAggregate

open override val parallelism: Int

open override val scheduler: Scheduler

Functions

fun NamedAggregate.aggregateId(id: String = generateId(), tenantId: String = TenantId.DEFAULT_TENANT_ID): DefaultAggregateId

Creates an AggregateId for this NamedAggregate with the specified parameters.

fun <T> NamedAggregate.aggregateType(): Class<T>?

Finds the aggregate type class associated with this named aggregate.

asAggregateMetadata

fun <C : Any, S : Any> NamedAggregate.asAggregateMetadata(): AggregateMetadata<C, S>

Converts this NamedAggregate to its corresponding AggregateMetadata.

fun cancel()

open override fun close()

Closes the dispatcher by canceling the subscription.

open fun currentContext(): Context?

open fun dispose()

fun NamedAggregate.generateId(): String

Generates a unique ID string for this NamedAggregate.

getContextAlias

fun NamedBoundedContext.getContextAlias(boundedContext: BoundedContext? = MetadataSearcher.metadata.contexts[contextName]): String

getContextAliasPrefix

fun NamedBoundedContext.getContextAliasPrefix(): String

Gets the context alias prefix for this bounded context.

open override fun handleExchange(exchange: ServerCommandExchange<*>): Mono<Void>

Handles a single command exchange by setting up the processing context and delegating to the command handler.

fun NamedAggregate.isLocal(): Boolean

Checks if the named aggregate is available locally at runtime.

isSameAggregateName

open fun isSameAggregateName(other: NamedAggregate): Boolean

isSameBoundedContext

open fun isSameBoundedContext(other: NamedBoundedContext): Boolean

fun NamedAggregate.materialize(): MaterializedNamedAggregate

Materializes this NamedAggregate into a MaterializedNamedAggregate.

fun onComplete()

fun onError(t: Throwable?)

fun onNext(value: Void?)

fun onSubscribe(s: Subscription?)

fun request(n: Long)

requestUnbounded

fun requestUnbounded()

requiredAggregateType

fun <T> NamedAggregate.requiredAggregateType(): Class<T>

Finds the aggregate type class associated with this named aggregate, throwing an exception if not found.

open override fun run()

Starts the dispatcher by subscribing to the message flux.

open fun safeOnNext(value: Void)

Safely processes the next message.

safeOnNextError

open fun safeOnNextError(value: Void, throwable: Throwable)

Handles errors that occur during message processing.

toEventNamedAggregate

fun NamedAggregate.toEventNamedAggregate(eventName: String): EventNamedAggregate

Extension function to convert a NamedAggregate to an EventNamedAggregate.

open override fun ServerCommandExchange<*>.toGroupKey(): Int

Generates a group key for the command exchange to ensure proper parallelism and ordering.

toNamedAggregateString

fun NamedAggregate.toNamedAggregateString(): String

Converts this NamedAggregate to its string representation.

toStringWithAlias

fun NamedAggregate.toStringWithAlias(): String

Converts this NamedAggregate to a string representation using the context alias.