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

InMemoryStateEventBus

class InMemoryStateEventBus(val sinkSupplier: (NamedAggregate) -> Sinks.Many<StateEvent<*>> = { Sinks.many().multicast().onBackpressureBuffer() }) : 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.many().multicast().onBackpressureBuffer() })

Properties

sinkSupplier

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

topicKind

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.

createExchange

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

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

metrizable

fun LocalStateEventBus.metrizable(): LocalStateEventBus

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

receive

open override fun receive(namedAggregates: Set<NamedAggregate>): Flux<StateEventExchange<*>>

Receives messages for the specified named aggregates.

send

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

Sends a message through the in-memory bus.

subscriberCount

open override fun subscriberCount(namedAggregate: NamedAggregate): Int

Returns the number of subscribers for the specified named aggregate.