EventStoreMessage

Outbound durable event message. Unlike EventMessage, events store messages are persisted and can be replayed.

msg = KubeMQ::PubSub::EventStoreMessage.new(
  channel: "orders.created",
  metadata: "order-event",
  body: '{"order_id": 100}',
  tags: { "region" => "us-east" }
)

Attributes

Attribute	Type	Default	Description
`id`	`String`	Auto-generated UUID	Unique message identifier
`channel`	`String`	Required	Target channel name
`metadata`	`String`	`nil`	Arbitrary metadata string
`body`	`String`	`nil`	Message payload (binary-safe)
`tags`	`Hash{String => String}`	`{}`	User-defined key-value tags

EventStoreStartPosition

Constants controlling where the broker begins replaying persisted events.

Constant	Value	Description
`START_NEW_ONLY`	`1`	Only events published after the subscription starts
`START_FROM_FIRST`	`2`	Replay from the first event ever stored
`START_FROM_LAST`	`3`	Replay from the most recent event
`START_AT_SEQUENCE`	`4`	Replay from a specific sequence number
`START_AT_TIME`	`5`	Replay from a specific Unix timestamp
`START_AT_TIME_DELTA`	`6`	Replay events stored within the last N seconds

EventsStoreSubscription

Configuration for subscribing to durable events store channels.

sub = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: "orders.created",
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_FIRST,
  start_position_value: 0,
  group: nil
)

Attributes

Attribute	Type	Default	Description
`channel`	`String`	Required	Channel name
`start_position`	`Integer`	Required	One of the `EventStoreStartPosition` constants
`start_position_value`	`Integer`	`0`	Sequence number, timestamp, or delta
`group`	`String`	`nil`	Consumer group for load-balanced delivery

PubSubClient Methods

`send_event_store(message)`

Sends a single event to a durable events store channel.

result = client.send_event_store(msg)
puts "Stored: #{result.id}" if result.sent

Returns: EventStoreResult

Raises: ValidationError, ClientClosedError, ConnectionError

`subscribe_to_events_store(subscription, cancellation_token:, on_error:, &block)`

Subscribes to durable events store messages. On reconnect, playback resumes from the last received sequence number.

sub = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: "orders",
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_FIRST
)
token = KubeMQ::CancellationToken.new

client.subscribe_to_events_store(sub, cancellation_token: token) do |event|
  puts "seq=#{event.sequence}: #{event.body}"
end

Yields: EventStoreReceived with channel, metadata, body, tags, id, and sequence attributes.

Returns: Subscription handle.

`create_events_store_sender`

Creates a streaming events store sender for high-throughput publishing with per-message confirmation.

sender = client.create_events_store_sender
result = sender.publish(msg)
puts "Sent: #{result.id}, confirmed: #{result.sent}"
sender.close

Returns: PubSub::EventStoreSender

EventStoreReceived

Inbound event received from an events store subscription.

Attribute	Type	Description
`id`	`String`	Message identifier
`channel`	`String`	Channel the event arrived on
`metadata`	`String`	Metadata string
`body`	`String`	Message payload
`tags`	`Hash`	Key-value tags
`sequence`	`Integer`	Broker-assigned sequence number

EventStoreResult

Result returned from send_event_store.

Attribute	Type	Description
`id`	`String`	Message identifier
`sent`	`Boolean`	Whether the event was persisted
`error`	`String`	Error message (if any)

Events Store

EventStoreMessage

Attributes

EventStoreStartPosition

EventsStoreSubscription

Attributes

PubSubClient Methods

`send_event_store(message)`

`subscribe_to_events_store(subscription, cancellation_token:, on_error:, &block)`

`create_events_store_sender`

EventStoreReceived

EventStoreResult

On this page