# Events (/connectors/stomp/how-to/events) Events are **fire-and-forget** pub/sub over the STOMP connector. SEND to a destination prefixed with `/topic/` and the connector routes the message to the KubeMQ **Events** pattern; every active subscriber on the matching channel gets a copy. Nothing is persisted — a subscriber that is offline at SEND time misses the message. ## Overview [#overview] The `/topic/` prefix selects the Events pattern; `/events/` is an accepted alias that egress always canonicalizes back to the primary `/topic/...` form — **lead with `/topic/` in your code**. The remaining destination segments are slash-to-dot joined into the KubeMQ channel: `/topic/orders/eu` → channel `orders.eu`. Events is the only pattern that accepts wildcard subscriptions and the default pattern for bare (prefixless) destinations. | Operation | STOMP action | KubeMQ mapping | | ------------------ | ------------------------------------ | --------------------------------------------------- | | Publish | `SEND /topic/` | `SendEvents` (`Store=false`) | | Subscribe | `SUBSCRIBE /topic/` (`ack:auto`) | Fan-out delivery to every matching subscriber | | Wildcard subscribe | `SUBSCRIBE /topic//*` or `/>` | The broker's native channel wildcards (Events only) | Events deliver as `ack:auto` — there is no client acknowledgement, no NACK, and no redelivery. A `receipt` on a SEND confirms only that **KubeMQ accepted the SEND**, not that any subscriber consumed it. ## How it works [#how-it-works] A published event fans out to **every** active subscriber whose filter matches. There is no consumer group and no load-balancing for Events — every matching subscriber receives every message. Use [Queues](/connectors/stomp/how-to/queues) when you need competing consumers. *Each event is copied to every subscriber whose filter matches the SENT destination; there is no persistence and no replay.* ## Publish and subscribe [#publish-and-subscribe] Each example subscribes **first** (Events have no replay — a SEND that beats the subscription is lost), waits briefly for the subscription to register, then publishes and drains the message. Every client reads the connector endpoint from `KUBEMQ_STOMP_URL` (default `tcp://localhost:61613`). ```go package main import ( "fmt" "log" "net/url" "os" "time" "github.com/go-stomp/stomp/v3" ) const destination = "/topic/demo" // Events pattern → channel demo 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. SUBSCRIBE FIRST — Events have no replay. 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) if err != nil { log.Fatalf("subscribe: %v", err) } time.Sleep(300 * time.Millisecond) // let the subscription register // 2. PUBLISH — SEND one event with a receipt (= KubeMQ accepted the SEND). pub, err := stomp.Dial(network, host) if err != nil { log.Fatalf("dial: %v", err) } if err := pub.Send(destination, "application/json", []byte(`{"message":"hello"}`), stomp.SendOpt.Receipt); err != nil { log.Fatalf("send: %v", err) } _ = pub.Disconnect() // 3. RECEIVE. select { case msg := <-subscription.C: fmt.Printf("received: %s (destination=%s)\n", string(msg.Body), msg.Destination) case <-time.After(10 * time.Second): log.Fatal("timed out waiting for the event") } } ``` ```python import os import queue from urllib.parse import urlparse import stomp DESTINATION = "/topic/demo" # Events pattern → channel demo 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. SUBSCRIBE FIRST — Events have no replay. listener = Listener() sub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000)) sub.set_listener("l", listener) sub.connect(wait=True) sub.subscribe(DESTINATION, id="demo", ack="auto") # 2. PUBLISH — SEND one event with a receipt (= KubeMQ accepted the SEND). pub = stomp.Connection12([(host, port)], heartbeats=(10000, 10000)) pub.connect(wait=True) pub.send(DESTINATION, '{"message":"hello"}', content_type="application/json") pub.disconnect() # 3. RECEIVE. frame = listener.inbox.get(timeout=10) print(f"received: {frame.body} (destination={frame.headers['destination']})") sub.disconnect() if __name__ == "__main__": main() ``` ```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/demo"; // Events pattern → channel demo 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. SUBSCRIBE FIRST — Events have no replay. BlockingQueue inbox = new ArrayBlockingQueue<>(1); StompSession sub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS); StompHeaders subHeaders = new StompHeaders(); subHeaders.setDestination(DESTINATION); subHeaders.setId("demo"); subHeaders.setAck("auto"); 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()); } }); Thread.sleep(300); // let the subscription register // 2. PUBLISH — SEND one event. StompSession pub = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS); StompHeaders pubHeaders = new StompHeaders(); pubHeaders.setDestination(DESTINATION); pubHeaders.add("content-type", "application/json"); pub.send(pubHeaders, "{\"message\":\"hello\"}".getBytes()); Thread.sleep(300); pub.disconnect(); // 3. RECEIVE. String body = inbox.poll(10, TimeUnit.SECONDS); if (body == null) throw new IllegalStateException("timed out waiting for the event"); System.out.printf("received: %s%n", body); sub.disconnect(); client.stop(); } } ``` ```typescript import { connect, type Client } from "stompit"; const DESTINATION = "/topic/demo"; // Events pattern → channel demo 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 { 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 { // 1. SUBSCRIBE FIRST — Events have no replay. const sub = await open(); const received = new Promise((resolve, reject) => { sub.subscribe({ destination: DESTINATION, ack: "auto" }, (err, message) => { if (err) return reject(err); message.readString("utf-8", (readErr, body) => (readErr ? reject(readErr) : resolve(body ?? ""))); }); }); await new Promise((r) => setTimeout(r, 300)); // let the subscription register // 2. PUBLISH — SEND one event. const pub = await open(); const frame = pub.send({ destination: DESTINATION, "content-type": "application/json" }); frame.write(JSON.stringify({ message: "hello" })); frame.end(); await new Promise((r) => pub.disconnect(() => r())); // 3. RECEIVE. console.log("received:", await received); await new Promise((r) => sub.disconnect(() => r())); } main().catch((err) => { console.error(err); process.exit(1); }); ``` ```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/demo"; // Events pattern → channel demo string brokerUri = $"stomp:tcp://{uri.Host}:{(uri.Port > 0 ? uri.Port : 61613)}"; var factory = new ConnectionFactory(brokerUri) { UserName = "kubemq", Password = "" }; // 1. SUBSCRIBE FIRST — Events have no replay. using var subConn = factory.CreateConnection(); subConn.Start(); using var subSession = subConn.CreateSession(AcknowledgementMode.IndividualAcknowledge); using var consumer = subSession.CreateConsumer(subSession.GetTopic(destination)); await Task.Delay(300); // let the subscription register // 2. PUBLISH — SEND one event. using (var pubConn = factory.CreateConnection()) { pubConn.Start(); using var pubSession = pubConn.CreateSession(AcknowledgementMode.AutoAcknowledge); using var producer = pubSession.CreateProducer(pubSession.GetTopic(destination)); var msg = producer.CreateBytesMessage(Encoding.UTF8.GetBytes("{\"message\":\"hello\"}")); msg.StompType = "application/json"; producer.Send(msg); } // 3. RECEIVE. var received = consumer.Receive(TimeSpan.FromSeconds(10)) ?? throw new InvalidOperationException("timed out waiting for the event"); Console.WriteLine($"received: {Encoding.UTF8.GetString(received.Content)} " + $"(channel={received.StompDestination?.PhysicalName})"); ``` ```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/demo" # Events pattern → channel demo # 1. SUBSCRIBE FIRST — Events have no replay. inbox = Thread::Queue.new sub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" }) sub.subscribe(DESTINATION, id: "demo", ack: "auto") { |msg| inbox << msg } sleep 0.3 # let the subscription register # 2. PUBLISH — SEND one event. pub = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" }) pub.publish(DESTINATION, '{"message":"hello"}', "content-type" => "application/json") pub.close # 3. RECEIVE. msg = Timeout.timeout(10) { inbox.pop } puts "received: #{msg.body} (destination=#{msg.headers['destination']})" sub.close ``` ```rust use std::time::Duration; use async_stomp::client::Connector; use async_stomp::{AckMode, FromServer, ToServer}; use futures::{SinkExt, StreamExt}; const DESTINATION: &str = "/topic/demo"; // Events pattern → channel demo 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> { let (host, port) = host_port(); // 1. SUBSCRIBE FIRST — Events have no replay. let mut sub = Connector::builder() .server(format!("{host}:{port}")) .virtualhost(&host) .connect() .await?; sub.send(ToServer::Subscribe { destination: DESTINATION.into(), id: "demo".into(), ack: Some(AckMode::Auto), }.into()) .await?; tokio::time::sleep(Duration::from_millis(300)).await; // let the subscription register // 2. PUBLISH — SEND one event. let mut publisher = Connector::builder() .server(format!("{host}:{port}")) .virtualhost(&host) .connect() .await?; publisher .send(ToServer::Send { destination: DESTINATION.into(), transaction: None, headers: Some(vec![("content-type".into(), "application/json".into())]), body: Some(br#"{"message":"hello"}"#.to_vec()), }.into()) .await?; publisher.send(ToServer::Disconnect { receipt: None }.into()).await?; // 3. RECEIVE. let frame = tokio::time::timeout(Duration::from_secs(10), sub.next()) .await? .ok_or("stream closed")??; if let FromServer::Message { body, destination, .. } = frame.content { let payload = String::from_utf8_lossy(&body.unwrap_or_default()).into_owned(); println!("received: {payload} (destination={destination})"); } Ok(()) } ``` ## Wildcard subscriptions [#wildcard-subscriptions] Wildcard filters are accepted on the **Events pattern only**, **on SUBSCRIBE only**, and use **the message broker's native wildcard syntax**: | Wildcard | Matches | | -------- | -------------------------------------------------------------------------------------------------------------------------------------------- | | `*` | exactly **one** segment, any position — `/topic/orders/*` matches `/topic/orders/eu` and `/topic/orders/us`, but not `/topic/orders/eu/west` | | `>` | the **tail**, and **must be the final** token — `/topic/orders/>` matches `/topic/orders/eu` and `/topic/orders/eu/west` | Delivery carries the **concrete matched channel**, not the subscription filter: a `/topic/orders/*` subscriber receiving on channel `orders.eu` gets `destination:/topic/orders/eu`. Route on the delivered `destination`, never on your filter string. **Wildcards are Events-only, subscribe-only, and use the broker's native syntax — there is no MQTT `+`/`#`.** `*` matches exactly one segment, `>` matches the final tail and must be the last token. There is **no** MQTT-style `+`/`#` translation — `+` and `#` are treated as literal segment characters, not wildcards. A wildcard on a SEND, or on any non-Events pattern (queues / store / RPC), is rejected with `ERROR "invalid destination"` and the connection closes. A literal `.` in a destination segment is **lossy** (`/topic/a.b` and `/topic/a/b` both map to channel `a.b`, and egress always re-emits the slash form) — prefer slashes, avoid literal dots. ## At-most-once delivery [#at-most-once-delivery] Events have no acknowledgement channel, so there is no NACK and no requeue. If a subscriber's per-subscriber output buffer (default 100) is full when an event arrives, the connector **drops that single delivery for that single subscriber** — the connection stays alive and **other** subscribers are unaffected. If you need every message guaranteed at least once, use [Queues](/connectors/stomp/how-to/queues) (acked, requeued) or [Events Store](/connectors/stomp/how-to/events-store) (persisted, replayable). ## Related [#related]