wow-core/me.ahoo.wow.eventsourcing.state/InMemoryStateEventBus

InMemoryStateEventBus

class InMemoryStateEventBus(val sinkSupplier: (NamedAggregate) -> Sinks.Many<StateEvent<*>> = { Sinks.unsafe().many().multicast().onBackpressureBuffer<StateEvent<*>>().concurrent() }) : InMemoryMessageBus<StateEvent<*>, StateEventExchange<*>> , LocalStateEventBus

In-memory implementation of LocalStateEventBus for testing and development. Uses Reactor Sinks for message broadcasting within the same JVM instance. Messages are not persisted and are lost when the application restarts.

Parameters

sinkSupplier

Supplier for creating sinks for each named aggregate (default: multicast sink with backpressure buffer).

Constructors

InMemoryStateEventBus

constructor(sinkSupplier: (NamedAggregate) -> Sinks.Many<StateEvent<*>> = { Sinks.unsafe().many().multicast().onBackpressureBuffer<StateEvent<*>>().concurrent() })

Properties

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

open override val topicKind: TopicKind

The topic kind for state events.

Functions

open override fun close()

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

open override fun StateEvent<*>.createExchange(): StateEventExchange<*>

Creates a StateEventExchange from a StateEvent message. Wraps the state event in a SimpleStateEventExchange for processing.

fun LocalStateEventBus.metrizable(): LocalStateEventBus

Wraps a LocalStateEventBus with metrics collection capabilities. Returns a MetricLocalStateEventBus that collects metrics on state event operations.

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

Receives messages for the specified named aggregates.

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

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

runtimeReceiver

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

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

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

Sends a message through the in-memory bus.

sendIfSubscribed

open override fun sendIfSubscribed(message: StateEvent<*>): 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.