Start from First

Overview

A new consumer joining an Events Store channel usually needs more than what happens next — it needs everything that already happened. start_at: :start_from_first solves that by replaying the channel's complete stored history before switching to live delivery, so a service can rebuild its state from scratch instead of starting with a blank slate and hoping nothing important was missed.

Under the hood, the broker walks the store from the oldest retained sequence forward, streaming each event to your on_event callback in order, then hands off to live delivery of new events without a gap. You don't manage offsets or checkpoints yourself — the start position is set once, at subscription time, via start_at: :start_from_first.

Gotchas: on a long-lived channel this can mean replaying millions of events before anything new shows up, so it's the wrong choice for a consumer that only cares about "from now on" (use :start_new_only for that). Retention and expiration policies still apply — events already purged by TTL or max-count limits are gone and won't be replayed, so "full history" only means what the store still has.

Prerequisites

• KubeMQ server running on localhost:50000
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-events-store.start-from-first"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-es-first")

for i <- 1..3 do
  {:ok, _} = KubeMQ.Client.send_event_store(client,
    KubeMQ.EventStore.new(channel: channel, body: "Historical event #{i}"))
end

IO.puts("Stored 3 events")
parent = self()
counter = :counters.new(1, [:atomics])

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events_store(client, channel,
    start_at: :start_from_first,
    on_event: fn event ->
      :counters.add(counter, 1, 1)
      IO.puts("Replayed: #{event.body} (seq: #{event.sequence})")
      send(parent, :replayed)
    end
  )

for _ <- 1..3 do
  receive do
    :replayed -> :ok
  after
    5_000 -> :timeout
  end
end

IO.puts("Total replayed: #{:counters.get(counter, 1)}")
KubeMQ.Subscription.cancel(sub)
KubeMQ.Client.close(client)

How It Works

• :start_from_first replays every stored event from sequence 1 onwards
• After replay completes, new events published to the channel are also delivered
• :counters provides a thread-safe counter across concurrent callbacks

Related

• Start from Last
• Start New Only