AggregateDispatcher
Abstract dispatcher for handling message exchanges for a specific aggregate with graceful shutdown support.
This dispatcher provides a robust framework for processing message exchanges in parallel, with built-in metrics collection, error handling, and graceful shutdown capabilities. Message exchanges are grouped by key for parallel processing, ensuring ordered execution within each group while allowing concurrent processing across different groups.
Key features:
Parallel message processing with configurable parallelism level
Metrics collection for monitoring dispatcher performance
Graceful shutdown that waits for active tasks to complete
Error handling through SafeSubscriber integration
Scheduler-based execution for resource management
Example usage:
class CustomAggregateDispatcher(
private val receiver: MessageReceiver<CommandExchange>,
override val parallelism: Int = 4,
override val scheduler: Scheduler = Schedulers.boundedElastic(),
) : AggregateDispatcher<CommandExchange>(
messageReadiness = receiver.readiness,
processingAdmission = receiver::openProcessing,
processingQuiescence = receiver::closeProcessing,
) {
override val messageFlux: Flux<CommandExchange> = receiver.messages
override fun CommandExchange.toGroupKey(): Int {
return command.aggregateId.hashCode() % parallelism
}
override fun handleExchange(exchange: CommandExchange): Mono<Void> {
return commandHandler.handle(exchange)
.doOnSuccess { exchange.acknowledge() }
}
}
// Usage
val dispatcher = CustomAggregateDispatcher(commandBus.runtimeReceiver(subscription))
val runtime = WowRuntime(
components = listOf(dispatcher),
shutdownTimeout = Duration.ofSeconds(30),
shutdownQuietPeriod = Duration.ZERO,
)
runtime.start().block()
runtime.stopGracefully().block()Parameters
Bounded dispatcher used for detached physical cancellation.
Completion signal for asynchronous message-source initialization. The message flux is subscribed before this signal is awaited.
Explicit transport-processing gate opened by start after dispatcher demand opens.
Prompt logical transport gate closed before physical source cancellation is detached.
Instance-scoped metrics recorder for handled exchanges.
Type Parameters
The type of message exchange being handled, must implement MessageExchange
See also
for the interface this class implements
for error handling capabilities
for the exchange type contract
Inheritors
Properties
Functions
Handles a single message exchange.
Prepares the dispatcher by subscribing without requesting messages.
Performs a graceful shutdown of the dispatcher.
Converts a message exchange to a grouping key for parallel processing.