wow-core/me.ahoo.wow.eventsourcing.snapshot.dispatcher/AggregateSnapshotDispatcher

AggregateSnapshotDispatcher

class AggregateSnapshotDispatcher(val name: String = "-", val namedAggregate: NamedAggregate, val messageFlux: Flux<StateEventExchange<*>>, val parallelism: Int = MessageParallelism.DEFAULT_PARALLELISM, snapshotHandler: SnapshotHandler, val scheduler: Scheduler, messageReadiness: Mono<Void> = Mono.empty(), processingAdmission: () -> Unit = {}, processingQuiescence: () -> Unit = {}) : AggregateDispatcher<StateEventExchange<*>> , ProcessorInfo

Dispatcher for handling snapshot operations on state events for a specific aggregate. Routes state event exchanges to the snapshot handler for processing.

Parameters

name

the name of this dispatcher (default: aggregateName-AggregateSnapshotDispatcher)

namedAggregate

the named aggregate this dispatcher handles

messageFlux

the flux of state event exchanges to process

parallelism

the number of parallel processing groups (default: MessageParallelism.DEFAULT_PARALLELISM)

snapshotHandler

the handler responsible for creating and storing snapshots

scheduler

the scheduler for processing 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

Constructors

AggregateSnapshotDispatcher

constructor(name: String = "-", namedAggregate: NamedAggregate, messageFlux: Flux<StateEventExchange<*>>, parallelism: Int = MessageParallelism.DEFAULT_PARALLELISM, snapshotHandler: SnapshotHandler, scheduler: Scheduler, messageReadiness: Mono<Void> = Mono.empty(), processingAdmission: () -> Unit = {}, processingQuiescence: () -> Unit = {})

Properties

open override val aggregateName: String

open override val contextName: String

The context name of the aggregate.

open override val isDisposed: Boolean

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

open override val name: String

open override val namedAggregate: NamedAggregate

open override val parallelism: Int

open override val processorName: String

The processor name, set to SNAPSHOT_PROCESSOR_NAME.

open override val scheduler: Scheduler

terminatedSignal

open override val terminatedSignal: Mono<Void>

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 fun currentContext(): Context?

open fun dispose()

override fun forceStop()

Releases resources promptly without blocking, and remains safe before prepare and across repeated or overlapping calls.

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: StateEventExchange<*>): Mono<Void>

Handles a state event exchange by setting the snapshot function and delegating to the snapshot 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?)

override fun prepare(runtimeContext: RuntimeContext): Mono<Void>

Prepares the dispatcher by subscribing without requesting messages.

override fun quiesce()

Stops admitting new work after the runtime has atomically closed global admission.

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 fun safeOnNext(value: Void)

Safely processes the next message.

safeOnNextError

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

Handles errors that occur during message processing.

override fun start()

override fun stopGracefully(): Mono<Void>

Performs a graceful shutdown of the dispatcher.

toEventNamedAggregate

fun NamedAggregate.toEventNamedAggregate(eventName: String): EventNamedAggregate

Extension function to convert a NamedAggregate to an EventNamedAggregate.

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

Computes the group key for parallel processing based on the message.

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.