# Events Store (/connectors/mqtt/how-to/events-store) Events Store is **persistent** pub/sub over the MQTT connector. Publish to a topic prefixed with `store/` and the message is durably stored by KubeMQ before delivery. Over MQTT the subscription start position is **always `StartNewOnly`** — there is no historical replay, so a subscriber receives only messages published *after* its subscription becomes active. ## Overview [#overview] The `store/` prefix selects the Events Store pattern. As with [Events](/connectors/mqtt/how-to/events), `/` in the path maps to `.` in the channel (`store/audit/log` → channel `audit.log`), publish and subscribe work on any MQTT version at any QoS, and v5 User Properties round-trip as KubeMQ `Tags`. The difference from Events is **persistence**: messages are stored, but MQTT subscribers can only stream forward from the moment they subscribe. | Operation | MQTT action | KubeMQ mapping | | ------------------ | -------------------------------- | ------------------------------------------------------- | | Publish | `PUBLISH store/` (any QoS) | `SendEvents` (`Store=true`) — persisted before delivery | | Subscribe | `SUBSCRIBE store/` (any QoS) | Durable subscription, start position `StartNewOnly` | | Wildcard subscribe | — | **Not supported** (SUBACK `0xA2`) | | Tags (v5) | `PUBLISH` User Properties | KubeMQ message `Tags` (bidirectional) | | Aspect | Events | Events Store | | ------------------ | --------- | --------------------------------------------------------- | | Persistence | No | Yes — messages stored in KubeMQ | | Offline delivery | Missed | Delivered once a subscriber connects (after publish time) | | Replay over MQTT | N/A | Not available (StartNewOnly only) | | Topic prefix | `events/` | `store/` | | Wildcard subscribe | Supported | Not supported | ## How it works [#how-it-works] A publish to `store/` is persisted by KubeMQ, then delivered to active subscribers. The connector registers every MQTT subscription with the store at `StartNewOnly`, so messages already in the store at subscribe time are never streamed to that subscriber. *Messages are durably stored, but an MQTT subscriber streams only from its subscribe point forward — pre-subscribe messages are never replayed.* **Events Store over MQTT is always `StartNewOnly` — there is no historical replay.** The connector hard-codes the subscribe start position. Messages published before the subscription was established are never delivered over MQTT, no matter how many are stored. If your application needs replay (start-from-first, start-at-sequence, start-at-time, …), use the KubeMQ gRPC or REST API instead. **Retain is silently dropped.** Setting the RETAIN flag on a `store/` publish still returns PUBACK `0x00` but the message is discarded — not delivered and not stored. The only retain-related failure is a CONNACK `0x9A` for a Will-retain flag at CONNECT. Never set retain on a message to KubeMQ. ## Publish and subscribe [#publish-and-subscribe] Each example demonstrates `StartNewOnly`: it publishes one message **before** subscribing (which a fresh subscriber never receives), subscribes to `store/`, then publishes a second message **after** subscribing and confirms only the post-subscribe message arrives. Every client reads the broker endpoint from `KUBEMQ_MQTT_URL` (default `tcp://localhost:1883`). ```go package main import ( "context" "fmt" "log" "net" "os" "strings" "time" "github.com/eclipse/paho.golang/paho" "github.com/google/uuid" ) const topic = "store/demo/s" // store/ prefix -> Events Store; channel demo.s func brokerURL() string { if u := os.Getenv("KUBEMQ_MQTT_URL"); u != "" { return u } return "tcp://localhost:1883" } func tcpAddr(raw string) string { for _, pfx := range []string{"tcp://", "ws://", "tls://"} { if strings.HasPrefix(raw, pfx) { return raw[len(pfx):] } } return raw } func dial(ctx context.Context, addr, id string, onMsg func(paho.PublishReceived) (bool, error)) *paho.Client { conn, err := net.Dial("tcp", addr) if err != nil { log.Fatalf("dial: %v", err) } cfg := paho.ClientConfig{Conn: conn} if onMsg != nil { cfg.OnPublishReceived = []func(paho.PublishReceived) (bool, error){onMsg} } c := paho.NewClient(cfg) ack, err := c.Connect(ctx, &paho.Connect{ClientID: id, KeepAlive: 30, CleanStart: true}) if err != nil { log.Fatalf("connect: %v", err) } if ack.ReasonCode != 0 { log.Fatalf("CONNACK reason=0x%02X", ack.ReasonCode) } return c } func main() { ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) defer cancel() addr := tcpAddr(brokerURL()) sfx := uuid.NewString()[:8] // 1. PUBLISH before subscribing — this is stored but never replayed over MQTT. pub := dial(ctx, addr, "go-store-pub-"+sfx, nil) if _, err := pub.Publish(ctx, &paho.Publish{Topic: topic, QoS: 1, Payload: []byte("pre-subscribe")}); err != nil { log.Fatalf("pre-publish: %v", err) } time.Sleep(300 * time.Millisecond) // 2. SUBSCRIBE — the connector registers a StartNewOnly subscription. got := make(chan string, 1) sub := dial(ctx, addr, "go-store-sub-"+sfx, func(pr paho.PublishReceived) (bool, error) { got <- string(pr.Packet.Payload) return true, nil }) subAck, err := sub.Subscribe(ctx, &paho.Subscribe{ Subscriptions: []paho.SubscribeOptions{{Topic: topic, QoS: 1}}, }) if err != nil { log.Fatalf("subscribe: %v", err) } // Wildcards on store/ are rejected with 0xA2. if subAck.Reasons[0] > 2 { log.Fatalf("SUBACK reason=0x%02X", subAck.Reasons[0]) } time.Sleep(700 * time.Millisecond) // let the StartNewOnly subscription go live // 3. PUBLISH after subscribing — only this message is delivered. if _, err := pub.Publish(ctx, &paho.Publish{Topic: topic, QoS: 1, Payload: []byte("post-subscribe")}); err != nil { log.Fatalf("post-publish: %v", err) } select { case msg := <-got: fmt.Printf("received: %s (StartNewOnly — the pre-subscribe message was not replayed)\n", msg) case <-ctx.Done(): log.Fatal("timed out waiting for the post-subscribe message") } _ = pub.Disconnect(&paho.Disconnect{ReasonCode: 0}) _ = sub.Disconnect(&paho.Disconnect{ReasonCode: 0}) } ``` ```python import os import threading import time import paho.mqtt.client as mqtt from paho.mqtt.enums import CallbackAPIVersion TOPIC = "store/demo/s" # store/ prefix -> Events Store; channel demo.s def parse_url(url: str) -> tuple[str, int]: scheme, rest = url.split("://", 1) host, _, port = rest.rstrip("/").partition(":") return host, int(port) if port else {"tcp": 1883, "tls": 8883, "ws": 8083}[scheme] def make_client(client_id: str) -> mqtt.Client: return mqtt.Client(CallbackAPIVersion.VERSION2, client_id=client_id, protocol=mqtt.MQTTv5) def main() -> None: host, port = parse_url(os.environ.get("KUBEMQ_MQTT_URL", "tcp://localhost:1883")) # 1. PUBLISH before subscribing — stored but never replayed over MQTT. pub = make_client("py-store-pub") pub.connect(host, port, keepalive=30) pub.loop_start() time.sleep(0.3) pub.publish(TOPIC, payload=b"pre-subscribe", qos=1).wait_for_publish(10) time.sleep(0.3) # 2. SUBSCRIBE — the connector registers a StartNewOnly subscription. received = threading.Event() payload: list[bytes] = [] sub = make_client("py-store-sub") sub.on_connect = lambda c, *_: c.subscribe(TOPIC, qos=1) sub.on_message = lambda c, u, m: (payload.append(m.payload), received.set()) sub.connect(host, port, keepalive=30) sub.loop_start() time.sleep(0.7) # let the StartNewOnly subscription go live # 3. PUBLISH after subscribing — only this message is delivered. pub.publish(TOPIC, payload=b"post-subscribe", qos=1).wait_for_publish(10) if not received.wait(timeout=10): raise TimeoutError("timed out waiting for the post-subscribe message") print(f"received: {payload[0].decode()!r} " "(StartNewOnly — the pre-subscribe message was not replayed)") pub.loop_stop(); pub.disconnect() sub.loop_stop(); sub.disconnect() if __name__ == "__main__": main() ``` ```java import java.nio.charset.StandardCharsets; import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.eclipse.paho.mqttv5.client.MqttAsyncClient; import org.eclipse.paho.mqttv5.client.MqttCallback; import org.eclipse.paho.mqttv5.client.MqttConnectionOptions; import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse; import org.eclipse.paho.mqttv5.client.IMqttToken; import org.eclipse.paho.mqttv5.common.MqttException; import org.eclipse.paho.mqttv5.common.MqttMessage; import org.eclipse.paho.mqttv5.common.packet.MqttProperties; public final class Main { // store/ prefix -> Events Store; KubeMQ channel demo.s ('/' -> '.'). private static final String TOPIC = "store/demo/s"; public static void main(String[] args) throws Exception { String broker = System.getenv().getOrDefault("KUBEMQ_MQTT_URL", "tcp://localhost:1883"); MqttConnectionOptions opts = new MqttConnectionOptions(); opts.setCleanStart(true); opts.setKeepAliveInterval(30); // 1. PUBLISH before subscribing — stored but never replayed over MQTT. MqttAsyncClient pub = new MqttAsyncClient(broker, "java-store-pub-" + UUID.randomUUID().toString().substring(0, 8)); pub.connect(opts).waitForCompletion(10_000); pub.publish(TOPIC, msg("pre-subscribe")).waitForCompletion(5_000); Thread.sleep(300); // 2. SUBSCRIBE — the connector registers a StartNewOnly subscription. CountDownLatch received = new CountDownLatch(1); String[] body = {null}; MqttAsyncClient sub = new MqttAsyncClient(broker, "java-store-sub-" + UUID.randomUUID().toString().substring(0, 8)); sub.setCallback(new MqttCallback() { public void messageArrived(String t, MqttMessage m) { body[0] = new String(m.getPayload(), StandardCharsets.UTF_8); received.countDown(); } public void disconnected(MqttDisconnectResponse r) {} public void mqttErrorOccurred(MqttException e) {} public void deliveryComplete(IMqttToken t) {} public void connectComplete(boolean reconnect, String uri) {} public void authPacketArrived(int code, MqttProperties props) {} }); sub.connect(opts).waitForCompletion(10_000); IMqttToken subToken = sub.subscribe(TOPIC, 1); subToken.waitForCompletion(10_000); // Wildcards on store/ are rejected with 0xA2. if (subToken.getGrantedQos()[0] >= 0x80) { throw new IllegalStateException("subscription rejected"); } Thread.sleep(700); // let the StartNewOnly subscription go live // 3. PUBLISH after subscribing — only this message is delivered. pub.publish(TOPIC, msg("post-subscribe")).waitForCompletion(5_000); if (!received.await(10, TimeUnit.SECONDS)) { throw new IllegalStateException("timed out waiting for the post-subscribe message"); } System.out.printf("received: %s (StartNewOnly — the pre-subscribe message was not replayed)%n", body[0]); pub.disconnect().waitForCompletion(5_000); pub.close(); sub.disconnect().waitForCompletion(5_000); sub.close(); } private static MqttMessage msg(String body) { MqttMessage m = new MqttMessage(body.getBytes(StandardCharsets.UTF_8)); m.setQos(1); m.setRetained(false); // retain is silently dropped by the connector return m; } } ``` ```typescript import mqtt, { type MqttClient } from "mqtt"; const TOPIC = "store/demo/s"; // store/ prefix -> Events Store; channel demo.s function connect(url: string, clientId: string): Promise { return new Promise((resolve, reject) => { const client = mqtt.connect(url, { clientId, protocolVersion: 5, clean: true, keepalive: 30 }); client.once("connect", () => resolve(client)); client.once("error", reject); }); } function publish(client: MqttClient, payload: string): Promise { return new Promise((resolve, reject) => { client.publish(TOPIC, payload, { qos: 1 }, (err) => (err ? reject(err) : resolve())); }); } async function main(): Promise { const url = process.env["KUBEMQ_MQTT_URL"] ?? "tcp://localhost:1883"; // 1. PUBLISH before subscribing — stored but never replayed over MQTT. const publisher = await connect(url, "js-store-pub"); await publish(publisher, "pre-subscribe"); await new Promise((r) => setTimeout(r, 300)); // 2. SUBSCRIBE — the connector registers a StartNewOnly subscription. const subscriber = await connect(url, "js-store-sub"); const received = new Promise((resolve) => { subscriber.on("message", (_topic, payload) => resolve(payload.toString())); }); await new Promise((resolve, reject) => { subscriber.subscribe(TOPIC, { qos: 1 }, (err) => (err ? reject(err) : resolve())); }); await new Promise((r) => setTimeout(r, 700)); // let the subscription go live // 3. PUBLISH after subscribing — only this message is delivered. await publish(publisher, "post-subscribe"); console.log(`received: ${await received} ` + "(StartNewOnly — the pre-subscribe message was not replayed)"); await publisher.endAsync(); await subscriber.endAsync(); } main().catch((err) => { console.error(err); process.exit(1); }); ``` ```csharp using System.Text; using MQTTnet; using MQTTnet.Client; using MQTTnet.Protocol; static (string host, int port) Endpoint() { var url = Environment.GetEnvironmentVariable("KUBEMQ_MQTT_URL") ?? "tcp://localhost:1883"; foreach (var pfx in new[] { "tcp://", "tls://", "ws://" }) if (url.StartsWith(pfx, StringComparison.OrdinalIgnoreCase)) url = url[pfx.Length..]; var parts = url.TrimEnd('/').Split(':'); return (parts[0], parts.Length > 1 ? int.Parse(parts[1]) : 1883); } var (host, port) = Endpoint(); const string topic = "store/demo/s"; // store/ prefix -> Events Store; channel demo.s var factory = new MqttFactory(); MqttClientOptions Options(string id) => new MqttClientOptionsBuilder() .WithTcpServer(host, port) .WithClientId($"{id}-{Guid.NewGuid():N}"[..26]) .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500) .WithCleanSession(true) .Build(); async Task Publish(IMqttClient client, string body) { var result = await client.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic(topic) .WithPayload(Encoding.UTF8.GetBytes(body)) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .Build()); if (result.ReasonCode != MqttClientPublishReasonCode.Success) throw new Exception($"PUBACK reason: {result.ReasonCode}"); } // 1. PUBLISH before subscribing — stored but never replayed over MQTT. using var pub = factory.CreateMqttClient(); await pub.ConnectAsync(Options("csharp-store-pub")); await Publish(pub, "pre-subscribe"); await Task.Delay(300); // 2. SUBSCRIBE — the connector registers a StartNewOnly subscription. var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); using var sub = factory.CreateMqttClient(); sub.ApplicationMessageReceivedAsync += e => { received.TrySetResult(Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment)); return Task.CompletedTask; }; await sub.ConnectAsync(Options("csharp-store-sub")); var subResult = await sub.SubscribeAsync(new MqttClientSubscribeOptionsBuilder() .WithTopicFilter(f => f.WithTopic(topic).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)) .Build()); // Wildcards on store/ are rejected with 0xA2. if (subResult.Items.First().ResultCode > MqttClientSubscribeResultCode.GrantedQoS2) throw new Exception("subscription rejected"); await Task.Delay(700); // let the StartNewOnly subscription go live // 3. PUBLISH after subscribing — only this message is delivered. await Publish(pub, "post-subscribe"); using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); Console.WriteLine($"received: {await received.Task.WaitAsync(cts.Token)} " + "(StartNewOnly — the pre-subscribe message was not replayed)"); await pub.DisconnectAsync(); await sub.DisconnectAsync(); ``` ```ruby # The mqtt gem speaks MQTT 3.1.1. Events Store publish/subscribe works on any # version; over MQTT the start position is always StartNewOnly regardless of client. require "mqtt" require "uri" require "securerandom" require "timeout" uri = URI.parse(ENV.fetch("KUBEMQ_MQTT_URL", "tcp://localhost:1883")) conn = { host: uri.host, port: uri.port, ssl: uri.scheme == "tls" } topic = "store/demo/s" # store/ prefix -> Events Store; channel demo.s # 1. PUBLISH before subscribing — stored but never replayed over MQTT. MQTT::Client.connect(host: conn[:host], port: conn[:port], ssl: conn[:ssl], client_id: "ruby-store-pub-#{SecureRandom.hex(4)}", clean_session: true, keep_alive: 30) do |client| client.publish(topic, "pre-subscribe", false, 1) end sleep 0.3 # 2. SUBSCRIBE — the connector registers a StartNewOnly subscription. received = Queue.new sub_ready = Queue.new subscriber = Thread.new do MQTT::Client.connect(host: conn[:host], port: conn[:port], ssl: conn[:ssl], client_id: "ruby-store-sub-#{SecureRandom.hex(4)}", clean_session: true, keep_alive: 30) do |client| client.subscribe([topic, 1]) sub_ready.push(:ready) _topic, payload = client.get received.push(payload) end end sub_ready.pop sleep 0.7 # let the StartNewOnly subscription go live # 3. PUBLISH after subscribing — only this message is delivered. MQTT::Client.connect(host: conn[:host], port: conn[:port], ssl: conn[:ssl], client_id: "ruby-store-pub2-#{SecureRandom.hex(4)}", clean_session: true, keep_alive: 30) do |client| client.publish(topic, "post-subscribe", false, 1) end payload = Timeout.timeout(10) { received.pop } puts "received: #{payload.inspect} (StartNewOnly — the pre-subscribe message was not replayed)" subscriber.kill ``` ```rust use rumqttc::v5::mqttbytes::v5::Packet; use rumqttc::v5::mqttbytes::QoS; use rumqttc::v5::{AsyncClient, Event, MqttOptions}; use std::env; use tokio::sync::oneshot; use tokio::time::{timeout, Duration}; use uuid::Uuid; const TOPIC: &str = "store/demo/s"; // store/ prefix -> Events Store; channel demo.s fn parse_host_port(url: &str) -> (String, u16) { let stripped = url .trim_start_matches("tcp://") .trim_start_matches("tls://") .trim_start_matches("ws://"); let host_port = stripped.split('/').next().unwrap_or(stripped); let mut parts = host_port.splitn(2, ':'); let host = parts.next().unwrap_or("localhost").to_string(); let port: u16 = parts.next().and_then(|p| p.parse().ok()).unwrap_or(1883); (host, port) } #[tokio::main] async fn main() -> Result<(), Box> { let url = env::var("KUBEMQ_MQTT_URL").unwrap_or_else(|_| "tcp://localhost:1883".to_string()); let (host, port) = parse_host_port(&url); let sfx = &Uuid::new_v4().to_string()[..8]; // Single connection: publish-before-subscribe, subscribe, publish-after. let mut opts = MqttOptions::new(format!("rust-store-{sfx}"), &host, port); opts.set_keep_alive(Duration::from_secs(30)); let (client, mut eventloop) = AsyncClient::new(opts, 10); let (msg_tx, msg_rx) = oneshot::channel::(); tokio::spawn(async move { let mut msg = Some(msg_tx); loop { match eventloop.poll().await { Ok(Event::Incoming(Packet::Publish(p))) => { if let Some(tx) = msg.take() { let _ = tx.send(String::from_utf8_lossy(&p.payload).to_string()); } return; } Ok(_) => {} Err(e) => { eprintln!("event loop: {e}"); return; } } } }); // 1. PUBLISH before subscribing — stored but never replayed over MQTT. client.publish(TOPIC, QoS::AtLeastOnce, false, b"pre-subscribe".as_ref()).await?; tokio::time::sleep(Duration::from_millis(400)).await; // 2. SUBSCRIBE — the connector registers a StartNewOnly subscription. client.subscribe(TOPIC, QoS::AtLeastOnce).await?; tokio::time::sleep(Duration::from_millis(700)).await; // let it go live // 3. PUBLISH after subscribing — only this message is delivered. client.publish(TOPIC, QoS::AtLeastOnce, false, b"post-subscribe".as_ref()).await?; let body = timeout(Duration::from_secs(10), msg_rx).await??; println!("received: {body} (StartNewOnly — the pre-subscribe message was not replayed)"); Ok(()) } ``` ## What Events Store does not have over MQTT [#what-events-store-does-not-have-over-mqtt] * **No historical replay.** The start position is always `StartNewOnly`; there is no MQTT topic syntax or v5 property to request a different one. * **No wildcard subscriptions.** A wildcard filter on a `store/` topic is rejected with SUBACK `0xA2` — wildcards are an [Events](/connectors/mqtt/how-to/events)-only feature. * **No durable resumption point.** Reconnecting a session does not resume from where it left off; you start fresh at `StartNewOnly` again. For any of these, drive Events Store through the KubeMQ gRPC or REST API, which exposes the full set of start positions. ## Related [#related]