Start from Last
Subscribe to a KubeMQ Events Store channel from the most recent stored event in Ruby to start with the latest persisted state.
Overview
A subscriber that just restarted usually doesn't need the entire event history — it needs to know where things stand right now without paying the cost of replaying everything that happened while it was offline. KubeMQ::PubSub::EventStoreStartPosition::START_FROM_LAST solves that: it re-anchors a new subscription to the tail of the store, delivering exactly one historical event (the most recently stored one) before switching to live delivery. That's the sweet spot between START_FROM_NEW (no history at all, so you might miss the current state entirely) and START_FROM_FIRST (the full backlog, which can be slow and mostly irrelevant for a consumer that only cares about "now").
Under the hood, start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_LAST is passed on the subscription. The broker looks up the channel's most recent stored event at subscription time, replays that single event to the new subscriber, and then streams every subsequently published event as it arrives — the same live path any other subscription uses.
Gotchas: if the channel is empty when you subscribe, there's no "last" event to deliver — you simply start receiving new events as they're published, with no error raised. START_FROM_LAST gives you one event, not the last N — if you need a short window of recent history, replay from a sequence number instead. And because "last" is resolved at subscribe time, two subscribers starting a few events apart can each get a different one.
Prerequisites
- KubeMQ server running on
localhost:50000 - Ruby SDK installed (
gem install kubemq)
Code
require 'kubemq'
address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
channel = 'ruby-events-store.start-from-last'
begin
client = KubeMQ::PubSubClient.new(address: address, client_id: 'es-last-example')
puts "Connected to #{address}"
3.times do |i|
msg = KubeMQ::PubSub::EventStoreMessage.new(channel: channel, metadata: "stored #{i}", body: "data-#{i}")
client.send_event_store(msg)
end
puts 'Pre-stored 3 events'
sleep 1
cancel = KubeMQ::CancellationToken.new
sub = KubeMQ::PubSub::EventsStoreSubscription.new(
channel: channel,
start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_LAST
)
client.subscribe_to_events_store(sub, cancellation_token: cancel, on_error: lambda { |e|
puts "Error: #{e.message}"
}) do |event|
puts "Received (from last): #{event.metadata}"
end
puts 'Subscribed with START_FROM_LAST — only the last stored + new events'
msg = KubeMQ::PubSub::EventStoreMessage.new(channel: channel, metadata: 'new after sub', body: 'new')
client.send_event_store(msg)
sleep 2
rescue KubeMQ::Error => e
puts "KubeMQ error: #{e.message}"
ensure
cancel&.cancel
client&.close
puts 'Done'
endHow It Works
START_FROM_LASTdelivers the most recent stored event, then continues with new events.- Use this when you only need the latest state, not the full history.
- Review timeouts, channel names, and client IDs before running against shared environments.
Related
Was this page helpful?