# Persistent Pub/Sub (/sdks/rust/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 an `EventStoreResult` confirming storage plus a broker-assigned sequence number, and `subscribe_to_events_store` takes a required `EventsStoreSubscription` variant telling the broker where to start — new events only, from the first stored event (`EventsStoreSubscription::StartFromFirst`, used here), or a given sequence or time. Production subscribers usually resume from a saved checkpoint instead of replaying from the beginning.

**Gotchas:** replaying from the first event on every restart replays the whole log, which gets costly on a busy channel — track the last sequence you processed instead. Starting from new-only has the opposite risk: anything published earlier is silently skipped, so don't rely on a fixed `sleep` like this sample's to paper over that race in production. 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`
* Rust SDK installed (`cargo add kubemq`)

## Code [#code]

```rust title="main.rs"
use kubemq::prelude::*;
use kubemq::{EventStoreBuilder, EventsStoreSubscription};
use std::time::Duration;

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

    let channel = "rust-events-store.persistent-pubsub";

    let sub = client
        .subscribe_to_events_store(
            channel,
            "",
            EventsStoreSubscription::StartFromFirst,
            |event| {
                Box::pin(async move {
                    println!(
                        "Received store event: id={}, seq={}, body={}",
                        event.id,
                        event.sequence,
                        String::from_utf8_lossy(&event.body)
                    );
                })
            },
            None,
        )
        .await?;

    tokio::time::sleep(Duration::from_millis(500)).await;

    let event = EventStoreBuilder::new()
        .channel(channel)
        .metadata("store-example")
        .body(b"Persistent message".to_vec())
        .build();

    let result = client.send_event_store(event).await?;
    println!("Event store sent: id={}, sent={}", result.id, result.sent);

    tokio::time::sleep(Duration::from_secs(2)).await;
    sub.unsubscribe().await;
    client.close().await?;
    Ok(())
}
```

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

* `EventStore` messages have `store: true` on the wire, enabling server-side persistence.
* `EventsStoreSubscription::StartFromFirst` replays all stored events before delivering new ones.
* `send_event_store` returns an `EventStoreResult` with the server-assigned ID and confirmation.
* Review timeouts, channel names, and client IDs before running against shared environments.
* Run the program while the server from the prerequisites is available.

## Related [#related]

* [Events Store overview](/learn/events-store)
* [Rust SDK Reference](/sdks/rust/reference/events-store)
* [Replay from Sequence](/sdks/rust/how-to/events-store/replay-from-sequence)
* [Start from First](/sdks/rust/how-to/events-store/start-from-first)