wow-core/me.ahoo.wow.metrics/MetricCommandBus

MetricCommandBus

open class MetricCommandBus<T : CommandBus>(delegate: T) : AbstractMetricDecorator<T> , CommandBus, Metrizable

Metric decorator for command buses that collects metrics on command sending and receiving operations. This class wraps any CommandBus implementation and adds metrics collection with tags for aggregate name, command name, and source identification.

Parameters

delegate

the underlying command bus implementation

Type Parameters

the specific type of CommandBus being decorated

Inheritors

MetricLocalCommandBus

MetricDistributedCommandBus

Constructors

MetricCommandBus

constructor(delegate: T)

Properties

delegate

override val delegate: T

the original component being decorated, must implement the Decorator interface

source

val source: String

The source identifier derived from the original delegate's class name. This is used for metrics tagging to identify the component type.

topicKind

open override val topicKind: TopicKind

The topic kind for command messages.

Functions

open override fun close()

Closes the command bus and releases any resources. This delegates to the underlying command bus implementation.

receive

open override fun receive(subscription: MessageSubscription): Flux<ServerCommandExchange<*>>

Receives command exchanges for the specified named aggregates and collects metrics on the operation. Metrics collected include timing and tags for aggregate identification and subscriber information.

receiver

open override fun receiver(subscription: MessageSubscription): MessageReceiver<ServerCommandExchange<*>>

Creates a single message source with an explicit transport readiness boundary.

runtimeReceiver

open override fun runtimeReceiver(subscription: MessageSubscription): MessageReceiver<ServerCommandExchange<*>>

Creates the message source owned by a me.ahoo.wow.runtime.WowRuntime dispatcher.

send

open override fun send(message: CommandMessage<*>): Mono<Void>

Sends a command message and collects metrics on the operation. Metrics collected include timing, success/failure rates, and tags for aggregate and command identification.

tagSource

fun <M : Any> Flux<M>.tagSource(): Flux<M>

Extension function to tag a Flux publisher with the current source identifier. This adds the source tag to metrics collected from this reactive stream.

fun <M : Any> Mono<M>.tagSource(): Mono<M>

Extension function to tag a Mono publisher with the current source identifier. This adds the source tag to metrics collected from this reactive stream.