wow-core/me.ahoo.wow.messaging.dispatcher/AggregateMessageDispatcher

AggregateMessageDispatcher

abstract class AggregateMessageDispatcher<T : MessageExchange<*, *>> : SafeSubscriber<Void> , MessageDispatcher, NamedAggregateDecorator

Abstract dispatcher for handling message exchanges for a specific aggregate.

This dispatcher groups message exchanges by a key for parallel processing, applies metrics, and handles each exchange on a specified scheduler.

Parameters

The type of message exchange being handled

Inheritors

AbstractAggregateEventDispatcher

AggregateSnapshotDispatcher

AggregateCommandDispatcher

Constructors

AggregateMessageDispatcher

constructor()

Types

Companion

object Companion

Properties

aggregateName

open override val aggregateName: String

contextName

open override val contextName: String

isDisposed

open override val isDisposed: Boolean

messageFlux

abstract val messageFlux: Flux<T>

The flux of message exchanges to be processed.

name

abstract val name: String

namedAggregate

abstract val namedAggregate: NamedAggregate

parallelism

abstract val parallelism: Int

The level of parallelism for processing grouped exchanges.

scheduler

abstract val scheduler: Scheduler

The scheduler to use for processing message exchanges.

Functions

aggregateId

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

Creates an AggregateId for this NamedAggregate with the specified parameters.

aggregateType

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.

cancel

fun cancel()

open override fun close()

Closes the dispatcher by canceling the subscription.

currentContext

open fun currentContext(): Context?

dispose

open fun dispose()

generateId

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.

handleExchange

abstract fun handleExchange(exchange: T): Mono<Void>

Handles a single message exchange.

isLocal

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

materialize

fun NamedAggregate.materialize(): MaterializedNamedAggregate

Materializes this NamedAggregate into a MaterializedNamedAggregate.

onComplete

fun onComplete()

onError

fun onError(t: Throwable?)

onNext

fun onNext(value: Void?)

onSubscribe

fun onSubscribe(s: Subscription?)

request

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.

run

open override fun run()

Starts the dispatcher by subscribing to the message flux.

safeOnNext

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.

toGroupKey

abstract fun T.toGroupKey(): Int

Converts a message exchange to a grouping key for parallel processing.

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.