# Events Store (/connectors/stomp/how-to/events-store)



Events Store is **persistent** pub/sub over the STOMP connector. SEND to a destination prefixed with `/topic-store/<channel>` and the connector routes the message to the KubeMQ **Events Store** pattern, which **persists** it. A later SUBSCRIBE can **replay*&#x2A; from a chosen position — the first message, the last, a specific sequence, a wall-clock time, or a relative window — using the &#x2A;*`start-from`*&#x2A; / &#x2A;*`start-value`** SUBSCRIBE headers.

## Overview [#overview]

The `/topic-store/` prefix selects the Events Store pattern; `/store/` is an accepted alias that egress always canonicalizes back to the primary `/topic-store/...` form — **lead with `/topic-store/` in your code**. (There is no `/topic_store/` or `/eventstore/` — those are not recognized prefixes.) The remaining segments are slash-to-dot joined into the KubeMQ channel: `/topic-store/orders/eu` → channel `orders.eu`.

Events Store is the persistent sibling of [Events](/connectors/stomp/how-to/events): same fan-out, same at-most-once **live** delivery, but messages are **durable** and a subscriber can **replay** history.

<Callout type="warn" title="No wildcards on Events Store">
  Unlike plain [Events](/connectors/stomp/how-to/events), **wildcards do not apply to Events Store**. A `*` or `>` in a `/topic-store/...` SUBSCRIBE destination → `ERROR "invalid destination"` and the connection closes. Subscribe to an **exact** `/topic-store/<channel>` and use the replay headers below to control history.
</Callout>

| Operation          | STOMP action                                                 | KubeMQ mapping                                    |
| ------------------ | ------------------------------------------------------------ | ------------------------------------------------- |
| Publish (persist)  | `SEND /topic-store/<ch>`                                     | `SendEvents` (`Store=true`) — message is stored   |
| Subscribe (live)   | `SUBSCRIBE /topic-store/<ch>`                                | `SubscribeEventsStore`, `new` (only new messages) |
| Subscribe (replay) | `SUBSCRIBE /topic-store/<ch>` + `start-from` / `start-value` | Replay from the store's start position            |

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

A SEND is persisted by the store; a SUBSCRIBE either takes only new messages (the default) or replays history from a chosen start position. Distinct replay positions on the same channel become independent subscriptions, so one consumer can replay from the beginning while another takes only new messages.

<Mermaid
  chart="`
graph LR
PUB[&#x22;Publisher<br/>(SEND /topic-store/audit)&#x22;]
CONN[&#x22;STOMP connector<br/>:61613&#x22;]
BROKER[&#x22;Message Broker&#x22;]
STORE[(&#x22;Persistence / Raft&#x22;)]
S1[&#x22;Subscriber A<br/>(start-from:first)&#x22;]
S2[&#x22;Subscriber B<br/>(start-from:new)&#x22;]

PUB -- &#x22;SEND&#x22; --> CONN
CONN -- &#x22;SendEvents (Store=true)&#x22; --> BROKER
BROKER --> STORE
STORE -. &#x22;replay (full history)&#x22; .-> S1
BROKER -. &#x22;only new&#x22; .-> S2

class PUB,S1,S2 client
class CONN connector
class BROKER broker
class STORE store
`"
/>

*A SEND is persisted; subscribers choose a start position — `first` replays the full history, `new` (the default) takes only messages from subscription time forward.*

## The `start-from` replay table [#the-start-from-replay-table]

Replay is requested at **SUBSCRIBE** time with two headers — `start-from` (the replay **mode**) and `start-value` (the replay **parameter**, required for some modes, forbidden for others):

| `start-from`   | `start-value`                                          | Replays from                    | Notes                                                               |
| -------------- | ------------------------------------------------------ | ------------------------------- | ------------------------------------------------------------------- |
| absent / `new` | must **not** be present                                | only **new** messages from now  | the default — same as plain Events live delivery                    |
| `first`        | must **not** be present                                | the **first** persisted message | full history from the beginning                                     |
| `last`         | must **not** be present                                | the **last** persisted message  | the most recent single message                                      |
| `sequence`     | **required*&#x2A;, numeric **≥ 0**                     | the given **sequence** number   | per-channel monotonic sequence                                      |
| `time`         | **required**, RFC3339 **or*&#x2A; unix-seconds **≥ 0** | the given **timestamp**         | both formats accepted (e.g. `2026-06-15T12:00:00Z` or `1750000000`) |
| `time-delta`   | **required*&#x2A;, numeric **> 0** (seconds)           | **now − delta** seconds         | a relative window, e.g. "last 60 seconds"                           |

A bad combination → `ERROR "invalid subscription"` and the connection closes, **before** the subscription registers. This includes a `start-value` present for `new` / `first` / `last` (those forbid a value), and `sequence` / `time` / `time-delta` with a missing, non-numeric, or out-of-range value. An unrecognized `start-from` → `ERROR "unknown start-from"` and close.

<Callout type="info">
  **Distinct replay positions are distinct subscriptions.** Two subscribers asking for **different** start positions on the same channel get **independent** subscriptions (unlike plain Events, which share per channel). This is what lets one consumer replay from `first` while another takes only `new`. Replay headers are **silently ignored** on a plain Events (`/topic/`) SUBSCRIBE — plain Events has no persistence to replay from.
</Callout>

## Persist and replay [#persist-and-replay]

Each example persists a batch of messages to a `/topic-store/...` destination, then subscribes with `start-from:first` and replays the full history in order. Every client reads the connector endpoint from `KUBEMQ_STOMP_URL` (default `tcp://localhost:61613`).

<Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    package main

    import (
    	"fmt"
    	"log"
    	"net/url"
    	"os"
    	"time"

    	"github.com/go-stomp/stomp/v3"
    )

    const destination = "/topic-store/audit" // Events Store pattern → channel audit
    const total = 3

    func addr() (network, host string) {
    	u, _ := url.Parse(os.Getenv("KUBEMQ_STOMP_URL"))
    	if u == nil || u.Host == "" {
    		return "tcp", "localhost:61613"
    	}
    	return "tcp", u.Host
    }

    func main() {
    	network, host := addr()

    	// 1. Persist — SEND `total` messages to the store.
    	pub, err := stomp.Dial(network, host)
    	if err != nil {
    		log.Fatalf("dial: %v", err)
    	}
    	for i := 1; i <= total; i++ {
    		if err := pub.Send(destination, "text/plain",
    			[]byte(fmt.Sprintf("evt-%d", i)), stomp.SendOpt.Receipt); err != nil {
    			log.Fatalf("send: %v", err)
    		}
    	}
    	_ = pub.Disconnect()

    	// 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
    	sub, err := stomp.Dial(network, host)
    	if err != nil {
    		log.Fatalf("dial: %v", err)
    	}
    	defer func() { _ = sub.Disconnect() }()
    	subscription, err := sub.Subscribe(destination, stomp.AckAuto,
    		stomp.SubscribeOpt.Header("start-from", "first"))
    	if err != nil {
    		log.Fatalf("subscribe: %v", err)
    	}

    	for got := 0; got < total; got++ {
    		select {
    		case msg := <-subscription.C:
    			fmt.Printf("replayed: %s\n", string(msg.Body))
    		case <-time.After(10 * time.Second):
    			log.Fatal("timed out replaying the store")
    		}
    	}
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    import os
    import queue
    from urllib.parse import urlparse

    import stomp

    DESTINATION = "/topic-store/audit"  # Events Store pattern → channel audit
    TOTAL = 3


    def endpoint() -> tuple[str, int]:
        parsed = urlparse(os.environ.get("KUBEMQ_STOMP_URL", "tcp://localhost:61613"))
        return parsed.hostname or "localhost", parsed.port or 61613


    class Listener(stomp.ConnectionListener):
        def __init__(self) -> None:
            self.inbox: queue.Queue = queue.Queue()

        def on_message(self, frame) -> None:
            self.inbox.put(frame)


    def main() -> None:
        host, port = endpoint()

        # 1. Persist — SEND `total` messages to the store.
        pub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000))
        pub.connect(wait=True)
        for i in range(1, TOTAL + 1):
            pub.send(DESTINATION, f"evt-{i}", content_type="text/plain")
        pub.disconnect()

        # 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
        listener = Listener()
        sub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000))
        sub.set_listener("l", listener)
        sub.connect(wait=True)
        sub.subscribe(DESTINATION, id="audit", ack="auto", headers={"start-from": "first"})

        for _ in range(TOTAL):
            frame = listener.inbox.get(timeout=10)
            print(f"replayed: {frame.body}")
        sub.disconnect()


    if __name__ == "__main__":
        main()
    ```
  </Tab>

  <Tab value="Java">
    ```java
    import java.lang.reflect.Type;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;

    import org.springframework.messaging.simp.stomp.StompHeaders;
    import org.springframework.messaging.simp.stomp.StompSession;
    import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter;
    import org.springframework.messaging.simp.stomp.ReactorNettyTcpStompClient;

    public final class Main {
        private static final String DESTINATION = "/topic-store/audit"; // → channel audit
        private static final int TOTAL = 3;

        public static void main(String[] args) throws Exception {
            String url = System.getenv().getOrDefault("KUBEMQ_STOMP_URL", "tcp://localhost:61613");
            java.net.URI u = java.net.URI.create(url);
            ReactorNettyTcpStompClient client = new ReactorNettyTcpStompClient(u.getHost(),
                    u.getPort() > 0 ? u.getPort() : 61613);

            // 1. Persist — SEND `total` messages to the store.
            StompSession pub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);
            for (int i = 1; i <= TOTAL; i++) {
                StompHeaders h = new StompHeaders();
                h.setDestination(DESTINATION);
                h.add("content-type", "text/plain");
                pub.send(h, ("evt-" + i).getBytes());
            }
            Thread.sleep(500);
            pub.disconnect();

            // 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
            BlockingQueue<String> inbox = new ArrayBlockingQueue<>(TOTAL);
            StompSession sub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS);
            StompHeaders subHeaders = new StompHeaders();
            subHeaders.setDestination(DESTINATION);
            subHeaders.setId("audit");
            subHeaders.setAck("auto");
            subHeaders.add("start-from", "first");
            sub.subscribe(subHeaders, new StompSessionHandlerAdapter() {
                @Override public Type getPayloadType(StompHeaders headers) { return String.class; }
                @Override public void handleFrame(StompHeaders headers, Object payload) {
                    inbox.add(payload == null ? "" : payload.toString());
                }
            });

            for (int i = 0; i < TOTAL; i++) {
                String body = inbox.poll(10, TimeUnit.SECONDS);
                if (body == null) throw new IllegalStateException("timed out replaying the store");
                System.out.printf("replayed: %s%n", body);
            }
            sub.disconnect();
            client.stop();
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    import { connect, type Client } from "stompit";

    const DESTINATION = "/topic-store/audit"; // Events Store pattern → channel audit
    const TOTAL = 3;

    function endpoint(): { host: string; port: number } {
      const url = new URL(process.env["KUBEMQ_STOMP_URL"] ?? "tcp://localhost:61613");
      return { host: url.hostname, port: Number(url.port) || 61613 };
    }

    function open(): Promise<Client> {
      const { host, port } = endpoint();
      return new Promise((resolve, reject) => {
        connect({ host, port, connectHeaders: { "accept-version": "1.2", "heart-beat": "10000,10000" } },
          (err, client) => (err ? reject(err) : resolve(client)));
      });
    }

    async function main(): Promise<void> {
      // 1. Persist — SEND `total` messages to the store.
      const pub = await open();
      for (let i = 1; i <= TOTAL; i++) {
        const frame = pub.send({ destination: DESTINATION, "content-type": "text/plain" });
        frame.write(`evt-${i}`);
        frame.end();
      }
      await new Promise<void>((r) => pub.disconnect(() => r()));

      // 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
      const sub = await open();
      let got = 0;
      await new Promise<void>((resolve, reject) => {
        const timer = setTimeout(() => reject(new Error("timed out replaying the store")), 15_000);
        sub.subscribe({ destination: DESTINATION, ack: "auto", "start-from": "first" }, (err, message) => {
          if (err) return reject(err);
          message.readString("utf-8", (readErr, body) => {
            if (readErr) return reject(readErr);
            console.log(`replayed: ${body}`);
            if (++got === TOTAL) {
              clearTimeout(timer);
              resolve();
            }
          });
        });
      });
      await new Promise<void>((r) => sub.disconnect(() => r()));
    }

    main().catch((err) => {
      console.error(err);
      process.exit(1);
    });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using System.Text;
    using Stomp.Net;

    var url = Environment.GetEnvironmentVariable("KUBEMQ_STOMP_URL") ?? "tcp://localhost:61613";
    var uri = new Uri(url);
    const string destination = "/topic-store/audit"; // Events Store pattern → channel audit
    const int total = 3;

    string brokerUri = $"stomp:tcp://{uri.Host}:{(uri.Port > 0 ? uri.Port : 61613)}";
    var factory = new ConnectionFactory(brokerUri) { UserName = "kubemq", Password = "" };

    // 1. Persist — SEND `total` messages to the store.
    using (var pubConn = factory.CreateConnection())
    {
        pubConn.Start();
        using var session = pubConn.CreateSession(AcknowledgementMode.AutoAcknowledge);
        using var producer = session.CreateProducer(session.GetTopic(destination));
        for (var i = 1; i <= total; i++)
        {
            var msg = producer.CreateBytesMessage(Encoding.UTF8.GetBytes($"evt-{i}"));
            msg.StompType = "text/plain";
            producer.Send(msg);
        }
    }

    // 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
    using var conn = factory.CreateConnection();
    conn.Start();
    using var consumeSession = conn.CreateSession(AcknowledgementMode.IndividualAcknowledge);
    // Stomp.Net forwards extra consumer headers, including the store replay directive.
    var topic = consumeSession.GetTopic(destination);
    using var consumer = consumeSession.CreateConsumer(topic, null, false,
        new Dictionary<string, string> { ["start-from"] = "first" });
    for (var i = 0; i < total; i++)
    {
        var msg = consumer.Receive(TimeSpan.FromSeconds(10))
            ?? throw new InvalidOperationException("timed out replaying the store");
        Console.WriteLine($"replayed: {Encoding.UTF8.GetString(msg.Content)}");
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    require "stomp"
    require "uri"
    require "timeout"

    uri = URI.parse(ENV.fetch("KUBEMQ_STOMP_URL", "tcp://localhost:61613"))
    hosts = [{ host: uri.host, port: uri.port || 61613, login: "kubemq", passcode: "" }]
    DESTINATION = "/topic-store/audit" # Events Store pattern → channel audit
    TOTAL = 3

    # 1. Persist — SEND `total` messages to the store.
    pub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" })
    (1..TOTAL).each { |i| pub.publish(DESTINATION, "evt-#{i}", "content-type" => "text/plain") }
    pub.close

    # 2. Replay — SUBSCRIBE start-from:first to read the full history in order.
    inbox = Thread::Queue.new
    sub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" })
    sub.subscribe(DESTINATION, id: "audit", ack: "auto", "start-from" => "first") { |msg| inbox << msg }

    TOTAL.times do
      msg = Timeout.timeout(10) { inbox.pop }
      puts "replayed: #{msg.body}"
    end
    sub.close
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    use std::time::Duration;

    use async_stomp::client::Connector;
    use async_stomp::{AckMode, FromServer, Message, ToServer};
    use futures::{SinkExt, StreamExt};

    const DESTINATION: &str = "/topic-store/audit"; // Events Store pattern → channel audit
    const TOTAL: usize = 3;

    fn host_port() -> (String, u16) {
        let url = std::env::var("KUBEMQ_STOMP_URL").unwrap_or_else(|_| "tcp://localhost:61613".into());
        let hp = url.trim_start_matches("tcp://").trim_start_matches("tls://");
        let mut parts = hp.splitn(2, ':');
        let host = parts.next().unwrap_or("localhost").to_string();
        let port = parts.next().and_then(|p| p.parse().ok()).unwrap_or(61613);
        (host, port)
    }

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn std::error::Error>> {
        let (host, port) = host_port();

        // 1. Persist — SEND `total` messages to the store.
        let mut pub_conn = Connector::builder()
            .server(format!("{host}:{port}"))
            .virtualhost(&host)
            .connect()
            .await?;
        for i in 1..=TOTAL {
            pub_conn
                .send(ToServer::Send {
                    destination: DESTINATION.into(),
                    transaction: None,
                    headers: Some(vec![("content-type".into(), "text/plain".into())]),
                    body: Some(format!("evt-{i}").into_bytes()),
                }.into())
                .await?;
        }
        pub_conn.send(ToServer::Disconnect { receipt: None }.into()).await?;

        // 2. Replay — SUBSCRIBE start-from:first via an extra header.
        let mut sub = Connector::builder()
            .server(format!("{host}:{port}"))
            .virtualhost(&host)
            .connect()
            .await?;
        let mut subscribe: Message<ToServer> = ToServer::Subscribe {
            destination: DESTINATION.into(),
            id: "audit".into(),
            ack: Some(AckMode::Auto),
        }.into();
        subscribe.extra_headers = vec![(b"start-from".to_vec(), b"first".to_vec())];
        sub.send(subscribe).await?;

        for _ in 0..TOTAL {
            let frame = tokio::time::timeout(Duration::from_secs(10), sub.next())
                .await?
                .ok_or("stream closed")??;
            if let FromServer::Message { body, .. } = frame.content {
                println!("replayed: {}", String::from_utf8_lossy(&body.unwrap_or_default()));
            }
        }
        Ok(())
    }
    ```
  </Tab>
</Tabs>

## Reliability [#reliability]

**Live** delivery on Events Store is **at-most-once**, exactly like plain Events: a full per-subscriber output buffer drops that one delivery and keeps the connection alive. Persistence guarantees the message is **stored** and **replayable** — it does not change the live delivery guarantee for an already-attached subscriber. Never promise exactly-once.

## Related [#related]

<Cards>
  <Card title="Events" href="/connectors/stomp/how-to/events" description="The ephemeral (non-persistent) sibling — fire-and-forget pub/sub with wildcard subscriptions." />

  <Card title="Queues" href="/connectors/stomp/how-to/queues" description="Competing-consumer work queues — acknowledged, at-least-once delivery." />

  <Card title="Destination grammar" href="/connectors/stomp/reference/destination-grammar" description="The full destination grammar, prefix-to-pattern mapping, and the store replay headers." />
</Cards>