EventMessage

Outbound event message for fire-and-forget pub/sub. Events are not persisted — use EventStoreMessage for durable delivery.

msg = KubeMQ::PubSub::EventMessage.new(
  channel: "notifications.email",
  metadata: "user-signup",
  body: '{"user_id": 42}',
  tags: { "priority" => "high" }
)

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

EventsSubscription

Configuration for subscribing to real-time events. Supports wildcard channels (* and >) and consumer groups.

sub = KubeMQ::PubSub::EventsSubscription.new(
  channel: "events.>",
  group: "workers"
)

Attributes

Attribute	Type	Default	Description
`channel`	`String`	Required	Channel name or wildcard pattern
`group`	`String`	`nil`	Consumer group for load-balanced delivery

PubSubClient Methods

`send_event(message)`

Sends a single event. Fire-and-forget semantics.

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

Returns: EventSendResult

Raises: ValidationError, ClientClosedError, ConnectionError

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

Subscribes to real-time events on a channel. Runs on a background thread with automatic reconnection.

token = KubeMQ::CancellationToken.new
sub = KubeMQ::PubSub::EventsSubscription.new(channel: "events.>")

subscription = client.subscribe_to_events(sub, cancellation_token: token) do |event|
  puts "#{event.channel}: #{event.body}"
end

token.cancel
subscription.wait(5)

Parameters:

Parameter	Type	Description
`subscription`	`EventsSubscription`	Channel and group configuration
`cancellation_token`	`CancellationToken`	Token for cooperative cancellation (auto-created if nil)
`on_error`	`Proc`	Callback receiving `Error` on stream failures
`&block`	Block	Called for each received event

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

Returns: Subscription handle to check status, cancel, or join.

`create_events_sender(on_error:)`

Creates a streaming event sender for high-throughput publishing.

sender = client.create_events_sender
sender.publish(msg)
sender.close

Returns: PubSub::EventSender

EventReceived

Inbound event received in a subscription callback.

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

EventSendResult

Result returned from send_event.

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

Events

On this page