wow-core/me.ahoo.wow.event/AggregateEventDispatcher

AggregateEventDispatcher

class AggregateEventDispatcher(val namedAggregate: NamedAggregate, val name: String = "-", val parallelism: Int = MessageParallelism.DEFAULT_PARALLELISM, val messageFlux: Flux<EventStreamExchange>, val functionRegistrar: MessageFunctionRegistrar<MessageFunction<Any, DomainEventExchange<*>, Mono<*>>>, val eventHandler: EventHandler, val scheduler: Scheduler) : AbstractAggregateEventDispatcher<EventStreamExchange>

Dispatcher for processing domain events within a specific aggregate context.

This class handles the distribution and processing of domain event streams for a particular named aggregate. It extends AbstractAggregateEventDispatcher to provide concrete implementation for event stream processing.

See also

AbstractAggregateEventDispatcher

EventStreamExchange

MessageFunctionRegistrar

Scheduler

Constructors

AggregateEventDispatcher

constructor(namedAggregate: NamedAggregate, name: String = "-", parallelism: Int = MessageParallelism.DEFAULT_PARALLELISM, messageFlux: Flux<EventStreamExchange>, functionRegistrar: MessageFunctionRegistrar<MessageFunction<Any, DomainEventExchange<*>, Mono<*>>>, eventHandler: EventHandler, scheduler: Scheduler)

Creates a new AggregateEventDispatcher with the specified parameters

Properties

open override val aggregateName: String

open override val contextName: String

open override val eventHandler: EventHandler

The handler for processing individual events

functionRegistrar

open override val functionRegistrar: MessageFunctionRegistrar<MessageFunction<Any, DomainEventExchange<*>, Mono<*>>>

The registrar containing event processing functions

open override val isDisposed: Boolean

open override val messageFlux: Flux<EventStreamExchange>

The flux of event stream exchanges to process

open override val name: String

The name of this dispatcher (default: derived from aggregate name)

open override val namedAggregate: NamedAggregate

The named aggregate this dispatcher handles

open override val parallelism: Int

The level of parallelism for processing (default: DEFAULT_PARALLELISM)

open override val scheduler: Scheduler

The scheduler for managing event processing concurrency

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.

createEventExchange

open override fun EventStreamExchange.createEventExchange(event: DomainEvent<*>): DomainEventExchange<*>

Creates a domain event exchange from an event stream exchange and domain event.

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

Handles a message exchange by processing all events in the stream.

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 EventStreamExchange.toGroupKey(): Int

Converts the exchange to a group 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.