wow-core/me.ahoo.wow.command/InMemoryCommandBus

InMemoryCommandBus

class InMemoryCommandBus(val sinkSupplier: (NamedAggregate) -> Sinks.Many<CommandMessage<*>> = { mpscUnicastManySink() }) : InMemoryMessageBus<CommandMessage<*>, ServerCommandExchange<*>> , LocalCommandBus

In-memory implementation of CommandBus for local command processing. This bus uses unicast sinks to ensure each command has exactly one consumer, making it suitable for single-instance or testing scenarios.

Author

ahoo wang

Parameters

sinkSupplier

Function that creates a unicast sink for each named aggregate. Defaults to an atomic MPSC unicast sink with backpressure buffer.

Constructors

InMemoryCommandBus

constructor(sinkSupplier: (NamedAggregate) -> Sinks.Many<CommandMessage<*>> = { mpscUnicastManySink() })

Properties

sinkSupplier

open override val sinkSupplier: (NamedAggregate) -> Sinks.Many<CommandMessage<*>>

Supplier for creating sinks for command distribution. Uses atomic MPSC admission and unicast mode to support concurrent senders while ensuring each command reaches exactly one consumer.

topicKind

open override val topicKind: TopicKind

The topic kind for command messages.

Functions

open override fun close()

Closes the message bus and releases any resources. Default implementation does nothing.

createExchange

open override fun CommandMessage<*>.createExchange(): ServerCommandExchange<*>

Creates a server command exchange for the given command message. This exchange handles the command processing lifecycle.

metrizable

fun LocalCommandBus.metrizable(): LocalCommandBus

Wraps a LocalCommandBus with metrics collection capabilities. Returns a MetricLocalCommandBus that collects metrics on command operations.

receive

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

Receives messages for the specified named aggregates.

receiver

open 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 message through the in-memory bus.

sendIfSubscribed

open override fun sendIfSubscribed(message: CommandMessage<*>): Mono<Boolean>

Attempts local delivery only while a processing subscriber is routable.

subscriberCount

open override fun subscriberCount(namedAggregate: NamedAggregate): Int

Returns the number of subscribers for the specified named aggregate.