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

AggregateSnapshotDispatcher

class AggregateSnapshotDispatcher(snapshotHandler: SnapshotHandler, val namedAggregate: NamedAggregate, val name: String = "-", val parallelism: Int = MessageParallelism.DEFAULT_PARALLELISM, val scheduler: Scheduler, val messageFlux: Flux<StateEventExchange<*>>) : AggregateMessageDispatcher<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

snapshotHandler

the handler responsible for creating and storing snapshots

namedAggregate

the named aggregate this dispatcher handles

name

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

parallelism

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

scheduler

the scheduler for processing messages

messageFlux

the flux of state event exchanges to process

Constructors

AggregateSnapshotDispatcher

constructor(snapshotHandler: SnapshotHandler, namedAggregate: NamedAggregate, name: String = "-", parallelism: Int = MessageParallelism.DEFAULT_PARALLELISM, scheduler: Scheduler, messageFlux: Flux<StateEventExchange<*>>)

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

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: 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?)

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