PubSubClient exposes publishEventStore, publishEventStoreStream, and subscribeToEventsStore for persisted events with configurable replay positions.

publishEventStore

suspend fun publishEventStore(message: EventStoreMessage): EventSendResult

Persists an event and returns acknowledgement metadata.

Parameters:

Name	Type	Required	Description
`message`	`EventStoreMessage`	Yes	Built via `eventStoreMessage { }` DSL

Returns: EventSendResult -- IDs, timestamps, and sent flag from the broker.

publishEventStoreStream

fun publishEventStoreStream(events: Flow<EventStoreMessage>): Flow<EventSendResult>

Streams persistent events for high-throughput publishing.

subscribeToEventsStore

fun subscribeToEventsStore(config: EventsStoreSubscriptionConfig.() -> Unit): Flow<EventMessageReceived>

Returns a cold Flow that subscribes with replay parameters.

Configuration DSL:

Property	Type	Required	Description
`channel`	`String`	Yes	Channel to subscribe to
`group`	`String`	No	Consumer group
`startPosition`	`StartPosition`	Yes	Replay start position

Replay options (StartPosition)

Value	Description
`StartPosition.StartNewOnly`	Only new events after subscription
`StartPosition.StartFromFirst`	Replay from the very first event
`StartPosition.StartFromLast`	Last event plus new ones
`StartPosition.StartAtSequence(n)`	Replay from sequence number `n`
`StartPosition.StartAtTime(nanos)`	Replay from timestamp (nanos since epoch)
`StartPosition.StartAtTimeDelta(seconds)`	Replay from `seconds` ago

eventStoreMessage DSL

eventStoreMessage {
    channel = "events-store.orders"
    body = payload.toByteArray()
    metadata = "order-service"
}

Quick Usage

EventsStore.kt

KubeMQClient.pubSub {
    address = "localhost:50000"
    clientId = "store-demo"
}.use { pub ->
    pub.publishEventStore(eventStoreMessage {
        channel = "audit.security"
        body = payload.toByteArray()
    })
}

Events Store