Events Store supports six subscription start positions that control where a subscriber begins reading from the event stream. This tutorial demonstrates each replay strategy with practical examples.

Subscription Start Positions

Each start position drops a subscriber at a different point in the stored stream: StartFromFirst rewinds to seq=1, StartAtSequence resumes at a chosen offset, StartFromLast catches the latest, and StartNewOnly ignores history and waits for new events.

Start Position	Enum Value	Description
`StartNewOnly`	1	Only events published after subscribing
`StartFromFirst`	2	Replay all events from the beginning
`StartFromLast`	3	Start from the most recent stored event
`StartAtSequence`	4	Start from a specific sequence number
`StartAtTime`	5	Start from a specific timestamp (Unix nanoseconds)
`StartAtTimeDelta`	6	Start from N seconds ago

Prerequisites

KubeMQ server running on localhost:50000
SDK installed (Getting Started)

Step-by-Step

Seed the Event Stream

Publish 10 order events to create a history for replay.

seed_orders.go

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/kubemq-io/kubemq-go/v2"
)

func main() {
    ctx := context.Background()
    client, err := kubemq.NewClient(ctx,
        kubemq.WithAddress("localhost", 50000),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    for i := 1; i <= 10; i++ {
        body := fmt.Sprintf(`{"orderId":"ORD-%04d","status":"created","total":%.2f}`,
            i, float64(i)*29.99)
        result, err := client.SendEventStore(ctx, kubemq.NewEvent().
            SetChannel("orders.events").
            SetBody([]byte(body)),
        )
        if err != nil {
            log.Fatal(err)
        }
        log.Printf("Stored seq=%s: ORD-%04d", result.EventID, i)
        time.Sleep(1 * time.Second)
    }
}

seed_orders.py

import json
import time
from kubemq import PubSubClient, EventStoreMessage

with PubSubClient(address="localhost:50000") as client:
    for i in range(1, 11):
        body = json.dumps({"orderId": f"ORD-{i:04d}", "status": "created",
                           "total": round(i * 29.99, 2)})
        result = client.publish_event_store(
            EventStoreMessage(channel="orders.events", body=body.encode("utf-8"))
        )
        print(f"Stored: ORD-{i:04d}")
        time.sleep(1)

seed_orders.ts

import { KubeMQClient, createEventStoreMessage } from 'kubemq-js';

const client = await KubeMQClient.create({ address: 'localhost:50000' });

for (let i = 1; i <= 10; i++) {
  await client.sendEventStore(
    createEventStoreMessage({
      channel: 'orders.events',
      body: JSON.stringify({
        orderId: `ORD-${String(i).padStart(4, '0')}`,
        status: 'created',
        total: +(i * 29.99).toFixed(2),
      }),
    })
  );
  console.log(`Stored: ORD-${String(i).padStart(4, '0')}`);
  await new Promise((r) => setTimeout(r, 1000));
}

SeedOrders.java

PubSubClient client = PubSubClient.builder()
    .address("localhost:50000")
    .clientId("seeder")
    .build();

for (int i = 1; i <= 10; i++) {
    String body = String.format(
        "{\"orderId\":\"ORD-%04d\",\"status\":\"created\",\"total\":%.2f}", i, i * 29.99);
    client.sendEventsStoreMessage(
        EventStoreMessage.builder()
            .channel("orders.events")
            .body(body.getBytes())
            .build());
    System.out.printf("Stored: ORD-%04d%n", i);
    Thread.sleep(1000);
}
client.close();

SeedOrders.cs

await using var client = new KubeMQClient(new KubeMQClientOptions());
await client.ConnectAsync();

for (var i = 1; i <= 10; i++)
{
    var body = $"{{\"orderId\":\"ORD-{i:D4}\",\"status\":\"created\",\"total\":{i * 29.99:F2}}}";
    await client.SendEventStoreAsync(new EventStoreMessage
    {
        Channel = "orders.events",
        Body = Encoding.UTF8.GetBytes(body),
    });
    Console.WriteLine($"Stored: ORD-{i:D4}");
    await Task.Delay(1000);
}

SeedOrders.kt

val client = KubeMQClient.pubSub {
    address = "localhost:50000"
    clientId = "seeder"
}

client.use {
    for (i in 1..10) {
        val body = """{"orderId":"ORD-${"%04d".format(i)}","status":"created","total":${"%.2f".format(i * 29.99)}}"""
        client.sendEventStore(eventStoreMessage {
            channel = "orders.events"
            this.body = body.toByteArray()
        })
        println("Stored: ORD-${"%04d".format(i)}")
        delay(1000)
    }
}

seed_orders.cc

kubemq::ClientOptions options;
options.set_address("localhost", 50000);
options.set_client_id("seeder");
auto client = kubemq::Client::Create(options).value();

for (int i = 1; i <= 10; ++i) {
    kubemq::EventStoreMessage msg;
    msg.set_channel("orders.events");
    msg.set_body("{\"orderId\":\"ORD-" + std::to_string(i) + "\",\"status\":\"created\"}");
    client->SendEventStore(msg);
    std::cout << "Stored: ORD-" << i << std::endl;
    std::this_thread::sleep_for(std::chrono::seconds(1));
}

seed_orders.rs

use kubemq::prelude::*;
use kubemq::EventStoreBuilder;
use std::time::Duration;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    for i in 1..=10 {
        let body = format!(
            r#"{{"orderId":"ORD-{:04}","status":"created","total":{:.2}}}"#,
            i,
            i as f64 * 29.99
        );
        let event = EventStoreBuilder::new()
            .channel("orders.events")
            .body(body.into_bytes())
            .build();
        let result = client.send_event_store(event).await?;
        println!("Stored seq=ORD-{:04}: sent={}", i, result.sent);
        tokio::time::sleep(Duration::from_secs(1)).await;
    }

    client.close().await?;
    Ok(())
}

seed_orders.rb

require 'kubemq'

client = KubeMQ::PubSubClient.new(address: 'localhost:50000', client_id: 'seeder')

(1..10).each do |i|
  body = { orderId: format('ORD-%04d', i), status: 'created',
           total: (i * 29.99).round(2) }.to_json
  result = client.send_event_store(
    KubeMQ::PubSub::EventStoreMessage.new(channel: 'orders.events', body: body)
  )
  puts "Stored: ORD-#{format('%04d', i)} sent=#{result.sent}"
  sleep 1
end

client.close

seed_orders.exs

{:ok, client} =
  KubeMQ.Client.start_link(address: "localhost:50000", client_id: "seeder")

for i <- 1..10 do
  body =
    Jason.encode!(%{
      orderId: "ORD-#{String.pad_leading(Integer.to_string(i), 4, "0")}",
      status: "created",
      total: Float.round(i * 29.99, 2)
    })

  {:ok, _} =
    KubeMQ.Client.send_event_store(
      client,
      KubeMQ.EventStore.new(channel: "orders.events", body: body)
    )

  IO.puts("Stored: ORD-#{String.pad_leading(Integer.to_string(i), 4, "0")}")
  Process.sleep(1_000)
end

KubeMQ.Client.close(client)

Replay from Beginning (StartFromFirst)

Receive every event ever stored in the channel, starting from sequence 1.

replay_from_first.go

sub, err := client.SubscribeToEventsStore(ctx, "orders.events", "",
    kubemq.StartFromFirst(),
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        fmt.Printf("[FromFirst] seq=%d body=%s\n",
            event.Sequence, string(event.Body))
    }),
    kubemq.WithOnError(func(err error) { log.Println(err) }),
)

replay_from_first.py

client.subscribe_to_events_store(
    subscription=EventsStoreSubscription(
        channel="orders.events",
        start_position=EventStoreStartPosition.StartFromFirst,
        on_receive_event_callback=lambda e: print(
            f"[FromFirst] seq={e.sequence} body={e.body.decode('utf-8')}"
        ),
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=CancellationToken(),
)

replay_from_first.ts

client.subscribeToEventsStore({
  channel: 'orders.events',
  startPosition: EventStoreStartPosition.StartFromFirst,
  onEvent: (msg) =>
    console.log(`[FromFirst] seq=${msg.sequence} body=${new TextDecoder().decode(msg.body)}`),
  onError: (err) => console.error(err.message),
});

ReplayFromFirst.java

client.subscribeToEventsStore(EventsStoreSubscription.builder()
    .channel("orders.events")
    .startPosition(EventStoreStartPosition.StartFromFirst)
    .onReceiveEventCallback(event ->
        System.out.printf("[FromFirst] seq=%d body=%s%n",
            event.getSequence(), new String(event.getBody())))
    .onErrorCallback(err -> System.err.println(err.getMessage()))
    .build());

ReplayFromFirst.cs

await foreach (var msg in client.SubscribeToEventsStoreAsync(
    new EventsStoreSubscription
    {
        Channel = "orders.events",
        StartPosition = EventStoreStartPosition.StartFromFirst,
    }))
{
    Console.WriteLine($"[FromFirst] seq={msg.Sequence} body={Encoding.UTF8.GetString(msg.Body.Span)}");
}

ReplayFromFirst.kt

client.subscribeToEventsStore {
    channel = "orders.events"
    startPosition = StartPosition.StartFromFirst
}.collect { msg ->
    println("[FromFirst] seq=${msg.sequence} body=${String(msg.body)}")
}

replay_from_first.cc

client->SubscribeToEventsStore("orders.events", "",
    kubemq::StartPosition::StartFromFirst,
    [](const kubemq::EventStoreReceived& msg) {
        std::cout << "[FromFirst] seq=" << msg.sequence()
                  << " body=" << msg.body() << std::endl;
    },
    [](const std::string& err) { std::cerr << err << std::endl; });

replay_from_first.rs

use kubemq::EventsStoreSubscription;

let sub = client
    .subscribe_to_events_store(
        "orders.events",
        "",
        EventsStoreSubscription::StartFromFirst,
        |event| {
            Box::pin(async move {
                println!(
                    "[FromFirst] seq={} body={}",
                    event.sequence,
                    String::from_utf8_lossy(&event.body)
                );
            })
        },
        None,
    )
    .await?;

replay_from_first.rb

cancel = KubeMQ::CancellationToken.new

sub = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: 'orders.events',
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_FIRST
)
client.subscribe_to_events_store(sub, cancellation_token: cancel,
                                 on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
  puts "[FromFirst] seq=#{event.sequence} body=#{event.body}"
end

replay_from_first.exs

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
    start_at: :start_from_first,
    on_event: fn event ->
      IO.puts("[FromFirst] seq=#{event.sequence} body=#{event.body}")
    end
  )

Output: receives all 10 events (seq 1-10).

Start from a Specific Sequence (StartAtSequence)

Resume from sequence number 7 to receive events 7-10 plus any new events.

replay_from_sequence.go

sub, err := client.SubscribeToEventsStore(ctx, "orders.events", "",
    kubemq.StartAtSequence(7),
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        fmt.Printf("[AtSeq7] seq=%d body=%s\n",
            event.Sequence, string(event.Body))
    }),
    kubemq.WithOnError(func(err error) { log.Println(err) }),
)

replay_from_sequence.py

client.subscribe_to_events_store(
    subscription=EventsStoreSubscription(
        channel="orders.events",
        start_position=EventStoreStartPosition.StartAtSequence,
        start_position_value=7,
        on_receive_event_callback=lambda e: print(
            f"[AtSeq7] seq={e.sequence} body={e.body.decode('utf-8')}"
        ),
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=CancellationToken(),
)

replay_from_sequence.ts

client.subscribeToEventsStore({
  channel: 'orders.events',
  startPosition: EventStoreStartPosition.StartAtSequence,
  startPositionValue: 7,
  onEvent: (msg) =>
    console.log(`[AtSeq7] seq=${msg.sequence} body=${new TextDecoder().decode(msg.body)}`),
  onError: (err) => console.error(err.message),
});

ReplayFromSequence.java

client.subscribeToEventsStore(EventsStoreSubscription.builder()
    .channel("orders.events")
    .startPosition(EventStoreStartPosition.StartAtSequence)
    .startPositionValue(7)
    .onReceiveEventCallback(event ->
        System.out.printf("[AtSeq7] seq=%d body=%s%n",
            event.getSequence(), new String(event.getBody())))
    .onErrorCallback(err -> System.err.println(err.getMessage()))
    .build());

ReplayFromSequence.cs

await foreach (var msg in client.SubscribeToEventsStoreAsync(
    new EventsStoreSubscription
    {
        Channel = "orders.events",
        StartPosition = EventStoreStartPosition.StartAtSequence,
        StartPositionValue = 7,
    }))
{
    Console.WriteLine($"[AtSeq7] seq={msg.Sequence} body={Encoding.UTF8.GetString(msg.Body.Span)}");
}

ReplayFromSequence.kt

client.subscribeToEventsStore {
    channel = "orders.events"
    startPosition = StartPosition.StartAtSequence
    startPositionValue = 7
}.collect { msg ->
    println("[AtSeq7] seq=${msg.sequence} body=${String(msg.body)}")
}

replay_from_sequence.cc

client->SubscribeToEventsStore("orders.events", "",
    kubemq::StartPosition::StartAtSequence, 7,
    [](const kubemq::EventStoreReceived& msg) {
        std::cout << "[AtSeq7] seq=" << msg.sequence()
                  << " body=" << msg.body() << std::endl;
    },
    [](const std::string& err) { std::cerr << err << std::endl; });

replay_from_sequence.rs

use kubemq::EventsStoreSubscription;

let sub = client
    .subscribe_to_events_store(
        "orders.events",
        "",
        EventsStoreSubscription::StartAtSequence(7),
        |event| {
            Box::pin(async move {
                println!(
                    "[AtSeq7] seq={} body={}",
                    event.sequence,
                    String::from_utf8_lossy(&event.body)
                );
            })
        },
        None,
    )
    .await?;

replay_from_sequence.rb

cancel = KubeMQ::CancellationToken.new

sub = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: 'orders.events',
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_AT_SEQUENCE,
  start_position_value: 7
)
client.subscribe_to_events_store(sub, cancellation_token: cancel,
                                 on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
  puts "[AtSeq7] seq=#{event.sequence} body=#{event.body}"
end

replay_from_sequence.exs

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
    start_at: {:start_at_sequence, 7},
    on_event: fn event ->
      IO.puts("[AtSeq7] seq=#{event.sequence} body=#{event.body}")
    end
  )

Output: receives events with seq 7, 8, 9, 10, then waits for new events.

Start from a Time Delta (StartAtTimeDelta)

Receive events published in the last 30 seconds.

replay_time_delta.go

sub, err := client.SubscribeToEventsStore(ctx, "orders.events", "",
    kubemq.StartAtTimeDelta(30),
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        fmt.Printf("[TimeDelta30s] seq=%d body=%s\n",
            event.Sequence, string(event.Body))
    }),
    kubemq.WithOnError(func(err error) { log.Println(err) }),
)

replay_time_delta.py

client.subscribe_to_events_store(
    subscription=EventsStoreSubscription(
        channel="orders.events",
        start_position=EventStoreStartPosition.StartAtTimeDelta,
        start_position_value=30,
        on_receive_event_callback=lambda e: print(
            f"[TimeDelta30s] seq={e.sequence} body={e.body.decode('utf-8')}"
        ),
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=CancellationToken(),
)

replay_time_delta.ts

client.subscribeToEventsStore({
  channel: 'orders.events',
  startPosition: EventStoreStartPosition.StartAtTimeDelta,
  startPositionValue: 30,
  onEvent: (msg) =>
    console.log(`[TimeDelta30s] seq=${msg.sequence} body=${new TextDecoder().decode(msg.body)}`),
  onError: (err) => console.error(err.message),
});

ReplayTimeDelta.java

client.subscribeToEventsStore(EventsStoreSubscription.builder()
    .channel("orders.events")
    .startPosition(EventStoreStartPosition.StartAtTimeDelta)
    .startPositionValue(30)
    .onReceiveEventCallback(event ->
        System.out.printf("[TimeDelta30s] seq=%d body=%s%n",
            event.getSequence(), new String(event.getBody())))
    .onErrorCallback(err -> System.err.println(err.getMessage()))
    .build());

ReplayTimeDelta.cs

await foreach (var msg in client.SubscribeToEventsStoreAsync(
    new EventsStoreSubscription
    {
        Channel = "orders.events",
        StartPosition = EventStoreStartPosition.StartAtTimeDelta,
        StartPositionValue = 30,
    }))
{
    Console.WriteLine($"[TimeDelta30s] seq={msg.Sequence} body={Encoding.UTF8.GetString(msg.Body.Span)}");
}

ReplayTimeDelta.kt

client.subscribeToEventsStore {
    channel = "orders.events"
    startPosition = StartPosition.StartAtTimeDelta
    startPositionValue = 30
}.collect { msg ->
    println("[TimeDelta30s] seq=${msg.sequence} body=${String(msg.body)}")
}

replay_time_delta.cc

client->SubscribeToEventsStore("orders.events", "",
    kubemq::StartPosition::StartAtTimeDelta, 30,
    [](const kubemq::EventStoreReceived& msg) {
        std::cout << "[TimeDelta30s] seq=" << msg.sequence()
                  << " body=" << msg.body() << std::endl;
    },
    [](const std::string& err) { std::cerr << err << std::endl; });

replay_time_delta.rs

use kubemq::EventsStoreSubscription;
use std::time::Duration;

let sub = client
    .subscribe_to_events_store(
        "orders.events",
        "",
        EventsStoreSubscription::StartAtTimeDelta(Duration::from_secs(30)),
        |event| {
            Box::pin(async move {
                println!(
                    "[TimeDelta30s] seq={} body={}",
                    event.sequence,
                    String::from_utf8_lossy(&event.body)
                );
            })
        },
        None,
    )
    .await?;

replay_time_delta.rb

cancel = KubeMQ::CancellationToken.new

sub = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: 'orders.events',
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_AT_TIME_DELTA,
  start_position_value: 30 # seconds
)
client.subscribe_to_events_store(sub, cancellation_token: cancel,
                                 on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
  puts "[TimeDelta30s] seq=#{event.sequence} body=#{event.body}"
end

replay_time_delta.exs

# The Elixir SDK expresses the time delta in milliseconds (30s = 30_000ms).
{:ok, sub} =
  KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
    start_at: {:start_at_time_delta, 30_000},
    on_event: fn event ->
      IO.puts("[TimeDelta30s] seq=#{event.sequence} body=#{event.body}")
    end
  )

Output: receives only events published within the last 30 seconds.

Start from the Last Event (StartFromLast)

Receive the most recently stored event, then all new events.

replay_from_last.go

sub, err := client.SubscribeToEventsStore(ctx, "orders.events", "",
    kubemq.StartFromLast(),
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        fmt.Printf("[FromLast] seq=%d body=%s\n",
            event.Sequence, string(event.Body))
    }),
    kubemq.WithOnError(func(err error) { log.Println(err) }),
)

replay_from_last.py

client.subscribe_to_events_store(
    subscription=EventsStoreSubscription(
        channel="orders.events",
        start_position=EventStoreStartPosition.StartFromLast,
        on_receive_event_callback=lambda e: print(
            f"[FromLast] seq={e.sequence} body={e.body.decode('utf-8')}"
        ),
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=CancellationToken(),
)

replay_from_last.ts

client.subscribeToEventsStore({
  channel: 'orders.events',
  startPosition: EventStoreStartPosition.StartFromLast,
  onEvent: (msg) =>
    console.log(`[FromLast] seq=${msg.sequence} body=${new TextDecoder().decode(msg.body)}`),
  onError: (err) => console.error(err.message),
});

ReplayFromLast.java

client.subscribeToEventsStore(EventsStoreSubscription.builder()
    .channel("orders.events")
    .startPosition(EventStoreStartPosition.StartFromLast)
    .onReceiveEventCallback(event ->
        System.out.printf("[FromLast] seq=%d body=%s%n",
            event.getSequence(), new String(event.getBody())))
    .onErrorCallback(err -> System.err.println(err.getMessage()))
    .build());

ReplayFromLast.cs

await foreach (var msg in client.SubscribeToEventsStoreAsync(
    new EventsStoreSubscription
    {
        Channel = "orders.events",
        StartPosition = EventStoreStartPosition.StartFromLast,
    }))
{
    Console.WriteLine($"[FromLast] seq={msg.Sequence} body={Encoding.UTF8.GetString(msg.Body.Span)}");
}

ReplayFromLast.kt

client.subscribeToEventsStore {
    channel = "orders.events"
    startPosition = StartPosition.StartFromLast
}.collect { msg ->
    println("[FromLast] seq=${msg.sequence} body=${String(msg.body)}")
}

replay_from_last.cc

client->SubscribeToEventsStore("orders.events", "",
    kubemq::StartPosition::StartFromLast,
    [](const kubemq::EventStoreReceived& msg) {
        std::cout << "[FromLast] seq=" << msg.sequence()
                  << " body=" << msg.body() << std::endl;
    },
    [](const std::string& err) { std::cerr << err << std::endl; });

replay_from_last.rs

use kubemq::EventsStoreSubscription;

let sub = client
    .subscribe_to_events_store(
        "orders.events",
        "",
        EventsStoreSubscription::StartFromLast,
        |event| {
            Box::pin(async move {
                println!(
                    "[FromLast] seq={} body={}",
                    event.sequence,
                    String::from_utf8_lossy(&event.body)
                );
            })
        },
        None,
    )
    .await?;

replay_from_last.rb

cancel = KubeMQ::CancellationToken.new

sub = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: 'orders.events',
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_FROM_LAST
)
client.subscribe_to_events_store(sub, cancellation_token: cancel,
                                 on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
  puts "[FromLast] seq=#{event.sequence} body=#{event.body}"
end

replay_from_last.exs

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
    start_at: :start_from_last,
    on_event: fn event ->
      IO.puts("[FromLast] seq=#{event.sequence} body=#{event.body}")
    end
  )

Output: receives event with seq=10 (the last stored), then waits for new events.

Start from a Specific Time (StartAtTime)

Receive events stored at or after a specific Unix timestamp in nanoseconds.

replay_from_time.go

targetTime := time.Now().Add(-5 * time.Minute).UnixNano()

sub, err := client.SubscribeToEventsStore(ctx, "orders.events", "",
    kubemq.StartAtTime(targetTime),
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        fmt.Printf("[AtTime] seq=%d body=%s\n",
            event.Sequence, string(event.Body))
    }),
    kubemq.WithOnError(func(err error) { log.Println(err) }),
)

replay_from_time.py

import time as time_mod

target_time = int((time_mod.time() - 300) * 1_000_000_000)  # 5 min ago in nanos

client.subscribe_to_events_store(
    subscription=EventsStoreSubscription(
        channel="orders.events",
        start_position=EventStoreStartPosition.StartAtTime,
        start_position_value=target_time,
        on_receive_event_callback=lambda e: print(
            f"[AtTime] seq={e.sequence} body={e.body.decode('utf-8')}"
        ),
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=CancellationToken(),
)

replay_from_time.ts

const targetTime = (Date.now() - 5 * 60 * 1000) * 1_000_000; // 5 min ago in nanos

client.subscribeToEventsStore({
  channel: 'orders.events',
  startPosition: EventStoreStartPosition.StartAtTime,
  startPositionValue: targetTime,
  onEvent: (msg) =>
    console.log(`[AtTime] seq=${msg.sequence} body=${new TextDecoder().decode(msg.body)}`),
  onError: (err) => console.error(err.message),
});

ReplayFromTime.java

long targetTime = (System.currentTimeMillis() - 300_000) * 1_000_000L; // 5 min ago

client.subscribeToEventsStore(EventsStoreSubscription.builder()
    .channel("orders.events")
    .startPosition(EventStoreStartPosition.StartAtTime)
    .startPositionValue(targetTime)
    .onReceiveEventCallback(event ->
        System.out.printf("[AtTime] seq=%d body=%s%n",
            event.getSequence(), new String(event.getBody())))
    .onErrorCallback(err -> System.err.println(err.getMessage()))
    .build());

ReplayFromTime.cs

var targetTime = (DateTimeOffset.UtcNow.AddMinutes(-5)).ToUnixTimeMilliseconds() * 1_000_000;

await foreach (var msg in client.SubscribeToEventsStoreAsync(
    new EventsStoreSubscription
    {
        Channel = "orders.events",
        StartPosition = EventStoreStartPosition.StartAtTime,
        StartPositionValue = targetTime,
    }))
{
    Console.WriteLine($"[AtTime] seq={msg.Sequence} body={Encoding.UTF8.GetString(msg.Body.Span)}");
}

ReplayFromTime.kt

val targetTime = (System.currentTimeMillis() - 300_000) * 1_000_000L

client.subscribeToEventsStore {
    channel = "orders.events"
    startPosition = StartPosition.StartAtTime
    startPositionValue = targetTime
}.collect { msg ->
    println("[AtTime] seq=${msg.sequence} body=${String(msg.body)}")
}

replay_from_time.cc

auto now = std::chrono::system_clock::now();
auto target = now - std::chrono::minutes(5);
auto nanos = std::chrono::duration_cast<std::chrono::nanoseconds>(
    target.time_since_epoch()).count();

client->SubscribeToEventsStore("orders.events", "",
    kubemq::StartPosition::StartAtTime, nanos,
    [](const kubemq::EventStoreReceived& msg) {
        std::cout << "[AtTime] seq=" << msg.sequence()
                  << " body=" << msg.body() << std::endl;
    },
    [](const std::string& err) { std::cerr << err << std::endl; });

replay_from_time.rs

use kubemq::EventsStoreSubscription;
use std::time::{Duration, SystemTime};

// The Rust SDK takes a SystemTime directly (5 minutes ago).
let target_time = SystemTime::now() - Duration::from_secs(5 * 60);

let sub = client
    .subscribe_to_events_store(
        "orders.events",
        "",
        EventsStoreSubscription::StartAtTime(target_time),
        |event| {
            Box::pin(async move {
                println!(
                    "[AtTime] seq={} body={}",
                    event.sequence,
                    String::from_utf8_lossy(&event.body)
                );
            })
        },
        None,
    )
    .await?;

replay_from_time.rb

cancel = KubeMQ::CancellationToken.new
# The Ruby SDK takes a Unix timestamp in seconds (5 minutes ago).
target_time = Time.now.to_i - 5 * 60

sub = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: 'orders.events',
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_AT_TIME,
  start_position_value: target_time
)
client.subscribe_to_events_store(sub, cancellation_token: cancel,
                                 on_error: ->(e) { puts "Error: #{e.message}" }) do |event|
  puts "[AtTime] seq=#{event.sequence} body=#{event.body}"
end

replay_from_time.exs

# The Elixir SDK takes a Unix timestamp in seconds (5 minutes ago).
target_time = System.system_time(:second) - 5 * 60

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
    start_at: {:start_at_time, target_time},
    on_event: fn event ->
      IO.puts("[AtTime] seq=#{event.sequence} body=#{event.body}")
    end
  )

Output: receives events stored at or after the target timestamp.

Choosing the Right Position

Scenario	Recommended Position	Why
Rebuild application state from scratch	`StartFromFirst`	Replays the entire history
Resume after a known checkpoint	`StartAtSequence`	Picks up exactly where you left off
Recover recent events after downtime	`StartAtTimeDelta`	Replays from a time window
Monitor live activity only	`StartNewOnly`	Ignores history, lowest overhead
Catch the latest event then go live	`StartFromLast`	Quick sync then real-time
Point-in-time recovery	`StartAtTime`	Precise timestamp-based replay

StartAtSequence requires a value greater than 0. StartAtTime requires a Unix timestamp in nanoseconds greater than 0. StartAtTimeDelta requires a positive number of seconds. Providing 0 or negative values results in a validation error.

Durable Replay Behavior

When a subscriber with a durable name reconnects, the start position parameter is ignored after the first connection. The store resumes from the last acknowledged sequence for that durable name. The durable name is {channel}-{group}.

To force a fresh replay, use a different group name or a different clientId.

Next Steps

Implement event sourcing using replay from first
Scale processing with consumer groups
Configure retention policies to manage storage
See Events Store Reference for all subscription parameters

Replay Events from Any Point