Events Store
Durable events with replay — KubeMQ Go SDK reference.
Events Store persists messages so subscribers can replay from a chosen position (first, last, sequence, time, or delta).
SendEventStore
func (c *Client) SendEventStore(ctx context.Context, event *EventStore) (*EventStoreResult, error)Sends a durable event and returns broker-assigned identifiers and timestamps.
Parameters:
| Name | Type | Required | Description |
|---|---|---|---|
ctx | context.Context | Yes | Send deadline |
event | *EventStore | Yes | Channel, body/metadata/tags for stored events |
Returns: *EventStoreResult — message ID and send metadata.
Throws: *KubeMQError — validation or transport errors.
SubscribeToEventsStore
func (c *Client) SubscribeToEventsStore(ctx context.Context, channel, group string, startOpt SubscriptionOption, opts ...SubscribeOption) (*Subscription, error)Subscribes with replay semantics. startOpt is a required start-position selector (StartFromNewEvents, StartFromFirstEvent, StartFromLastEvent, StartFromSequence, StartFromTime, or StartFromTimeDelta); opts are optional subscription behavior options.
Parameters:
| Name | Type | Required | Description |
|---|---|---|---|
ctx | context.Context | Yes | Subscription lifetime |
channel | string | Yes | Events Store channel |
group | string | No | Consumer group |
startOpt | SubscriptionOption | Yes | Replay start position (e.g. StartFromNewEvents()) |
opts | ...SubscribeOption | No | Buffers, callbacks, error handler |
Returns: *Subscription — unsubscribe handle.
Replay start positions
| Start | Behavior |
|---|---|
| Start from first | Replay from oldest retained message |
| Start from last | Begin after the latest message at subscribe time |
| Start new only | Only messages published after subscribe |
| Start at sequence | From a specific sequence number |
| Start at time / delta | Wall-clock or relative offsets |
Note:
startOptconstructors (StartFromNewEvents,StartFromSequence, etc.) are defined inevent_store.go;optsuseSubscribeOptionhelpers (WithOnEventStoreReceive,WithOnError, etc.). Align replay fields with server retention policies.
Quick Usage
es := kubemq.NewEventStore()
es.SetChannel("store.a")
es.SetBody([]byte("audit"))
res, err := client.SendEventStore(ctx, es)
if err != nil {
return err
}
_ = res.IdSee Also
Was this page helpful?