# Persistent Pub/Sub (/sdks/ruby/tutorials/persistent-pubsub)



## Overview [#overview]

This tutorial builds a publisher and subscriber on a KubeMQ Events Store channel — reach for this pattern when a subscriber can't guarantee it's listening the instant a message is published. Plain events are fire-and-forget: publish with no one subscribed and the message is gone. Events Store persists every event to a durable, ordered log, so a subscriber connecting seconds or a full restart later still catches up — useful for anything needing a complete history, like an audit trail or event-sourced state.

The two calls involved: `send_event_store` publishes and returns a result confirming storage with a `sent` flag plus a broker-assigned sequence number, and `subscribe_to_events_store` takes a subscription with a required `start_position` telling the broker where to start — new events only (`EventStoreStartPosition::START_NEW_ONLY`, used here), from the first stored event, or a given sequence or time. Production subscribers usually resume from a saved checkpoint instead of starting fresh.

**Gotchas:** starting from new events means anything published earlier is silently skipped — this sample papers over that race with a fixed `sleep` instead of a ready signal, fine for a demo but not production. Replaying from the first event on every restart replays the whole log, which gets costly on a busy channel. Persistence isn't consumer coordination: each independent subscriber gets its own full replay unless grouped with a consumer group.

## Prerequisites [#prerequisites]

* KubeMQ server running on `localhost:50000`
* Ruby SDK installed (`gem install kubemq`)

## Code [#code]

```ruby title="main.rb"
require 'kubemq'

address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
channel = 'ruby-events-store.persistent-pubsub'

begin
  client = KubeMQ::PubSubClient.new(address: address, client_id: 'es-basic-example')
  puts "Connected to #{address}"

  cancel = KubeMQ::CancellationToken.new

  sub = KubeMQ::PubSub::EventsStoreSubscription.new(
    channel: channel,
    start_position: KubeMQ::PubSub::EventStoreStartPosition::START_NEW_ONLY
  )
  client.subscribe_to_events_store(sub, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |event|
    puts "Received: channel=#{event.channel}, metadata=#{event.metadata}, body=#{event.body}"
  end
  sleep 1

  3.times do |i|
    msg = KubeMQ::PubSub::EventStoreMessage.new(
      channel: channel,
      metadata: "es event #{i}",
      body: "payload-#{i}"
    )
    result = client.send_event_store(msg)
    puts "Sent event #{i}: sent=#{result.sent}"
  end

  sleep 2
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  cancel&.cancel
  client&.close
  puts 'Done'
end
```

## How It Works [#how-it-works]

* Events store messages are persisted by the broker and can be replayed by future subscribers.
* `send_event_store` returns a confirmed result — unlike fire-and-forget events, the broker acknowledges persistence.
* The `start_position` controls where the subscription begins reading from the event log.
* Review timeouts, channel names, and client IDs before running against shared environments.

## Related [#related]

* [Pattern overview](/learn/events-store/getting-started)
* [Ruby SDK Reference](/sdks/ruby/reference)
* [Start from First](/sdks/ruby/how-to/events-store/start-from-first)
* [Replay from Sequence](/sdks/ruby/how-to/events-store/replay-from-sequence)