# Events (/connectors/mqtt/how-to/events) Events are **fire-and-forget** pub/sub over the MQTT connector. Publish to a topic prefixed with `events/` and the connector routes the message to the KubeMQ **Events** pattern; every matching subscriber gets a copy. Nothing is persisted — a subscriber that is offline at publish time misses the message. ## Overview [#overview] The `events/` prefix selects the Events pattern. Each path segment after the prefix is joined with `.` to form the KubeMQ channel: `events/site1/temp` → channel `site1.temp`. Publishing and subscribing both work on **any MQTT version** (3.1.1 or 5.0) at **any QoS**. Events is the only pattern that accepts wildcard subscriptions and the `DefaultPattern` for bare (prefixless) topics. | Operation | MQTT action | KubeMQ mapping | | ------------------ | -------------------------------------- | --------------------------------------------- | | Publish | `PUBLISH events/` (any QoS) | `SendEvents` (`Store=false`) | | Subscribe | `SUBSCRIBE events/` (any QoS) | Fan-out delivery to every matching subscriber | | Wildcard subscribe | `+` → one segment, `#` → trailing tail | KubeMQ `*` / `>` (Events only) | | Tags (v5) | `PUBLISH` User Properties | KubeMQ message `Tags` (bidirectional) | ## How it works [#how-it-works] A published event fans out to every connected subscriber whose filter matches. There is no consumer group and no load-balancing for Events — every matching subscriber receives every message. Use [Queues](/connectors/mqtt/how-to/queues) when you need competing consumers. *Each event is copied to every subscriber whose filter matches the published topic; there is no persistence and no replay.* **Retain is silently dropped.** The connector forces `RetainAvailable=0`. If you set the RETAIN flag on a runtime `PUBLISH`, the broker strips it: the `PUBACK` still returns `0x00` (success) but the message is **not delivered and not stored**, with no error. 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 subscribes **first** (Events have no replay — a publish that beats the subscription is lost), waits briefly for the subscription to register, then publishes and drains the message. On MQTT 5.0 connections, a publisher's User Properties round-trip as KubeMQ `Tags` and arrive back on the v5 subscriber. 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" ) // Subscribe with the single-level '+' wildcard; publish to a concrete leaf the // filter matches. '+' maps to KubeMQ '*'; '/' maps to '.' in the channel name. const subFilter = "events/demo/+" // KubeMQ channel filter demo.* const pubTopic = "events/demo/x" // KubeMQ channel demo.x 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(), 15*time.Second) defer cancel() addr := tcpAddr(brokerURL()) sfx := uuid.NewString()[:8] // 1. SUBSCRIBE FIRST — Events have no replay. got := make(chan string, 1) sub := dial(ctx, addr, "go-events-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: subFilter, QoS: 1}}, }) if err != nil { log.Fatalf("subscribe: %v", err) } // Reason codes > 2 are rejections (0xA2 = wildcard on a non-events pattern). if subAck.Reasons[0] > 2 { log.Fatalf("SUBACK reason=0x%02X", subAck.Reasons[0]) } time.Sleep(300 * time.Millisecond) // let the subscription register // 2. PUBLISH with a v5 User Property (round-trips as a KubeMQ Tag). pub := dial(ctx, addr, "go-events-pub-"+sfx, nil) pubAck, err := pub.Publish(ctx, &paho.Publish{ Topic: pubTopic, QoS: 1, Payload: []byte("hello"), Properties: &paho.PublishProperties{ User: []paho.UserProperty{{Key: "sensor", Value: "thermometer"}}, }, }) if err != nil { log.Fatalf("publish: %v", err) } if pubAck.ReasonCode != 0 { log.Fatalf("PUBACK reason=0x%02X", pubAck.ReasonCode) } // 3. RECEIVE. select { case msg := <-got: fmt.Printf("received: %s\n", msg) case <-ctx.Done(): log.Fatal("timed out waiting for event") } _ = 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 from paho.mqtt.properties import Properties from paho.mqtt.packettypes import PacketTypes SUB_FILTER = "events/demo/+" # '+' -> KubeMQ '*'; channel filter demo.* PUB_TOPIC = "events/demo/x" # KubeMQ channel demo.x 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 main() -> None: host, port = parse_url(os.environ.get("KUBEMQ_MQTT_URL", "tcp://localhost:1883")) received = threading.Event() payload: list[bytes] = [] # 1. SUBSCRIBE FIRST — Events have no replay. sub = mqtt.Client(CallbackAPIVersion.VERSION2, client_id="py-events-sub", protocol=mqtt.MQTTv5) sub.on_connect = lambda c, *_: c.subscribe(SUB_FILTER, 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.5) # let the subscription register before publishing # 2. PUBLISH with a v5 User Property (round-trips as a KubeMQ Tag). pub = mqtt.Client(CallbackAPIVersion.VERSION2, client_id="py-events-pub", protocol=mqtt.MQTTv5) pub.connect(host, port, keepalive=30) pub.loop_start() props = Properties(PacketTypes.PUBLISH) props.UserProperty = [("k1", "v1")] pub.publish(PUB_TOPIC, payload=b"hello", qos=1, properties=props).wait_for_publish(10) # 3. RECEIVE. if not received.wait(timeout=10): raise TimeoutError("timed out waiting for the event") print(f"received: {payload[0].decode()!r}") pub.loop_stop(); pub.disconnect() sub.loop_stop(); sub.disconnect() if __name__ == "__main__": main() ``` ```java import java.nio.charset.StandardCharsets; import java.util.List; 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; import org.eclipse.paho.mqttv5.common.packet.UserProperty; public final class Main { public static void main(String[] args) throws Exception { String broker = System.getenv().getOrDefault("KUBEMQ_MQTT_URL", "tcp://localhost:1883"); // '+' single-level wildcard -> KubeMQ '*'; '/' -> '.' in the channel name. String subFilter = "events/demo/+"; // channel filter demo.* String pubTopic = "events/demo/x"; // channel demo.x MqttConnectionOptions opts = new MqttConnectionOptions(); opts.setCleanStart(true); opts.setKeepAliveInterval(30); CountDownLatch received = new CountDownLatch(1); String[] body = {null}; // 1. SUBSCRIBE FIRST — Events have no replay. MqttAsyncClient sub = new MqttAsyncClient(broker, "java-events-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(subFilter, 1); subToken.waitForCompletion(10_000); // SUBACK >= 0x80 is a rejection (0xA2 = wildcard on a non-events pattern). if (subToken.getGrantedQos()[0] >= 0x80) { throw new IllegalStateException("subscription rejected"); } Thread.sleep(300); // let the subscription register // 2. PUBLISH with a v5 User Property (round-trips as a KubeMQ Tag). MqttAsyncClient pub = new MqttAsyncClient(broker, "java-events-pub-" + UUID.randomUUID().toString().substring(0, 8)); pub.connect(opts).waitForCompletion(10_000); MqttMessage msg = new MqttMessage("hello".getBytes(StandardCharsets.UTF_8)); msg.setQos(1); MqttProperties pubProps = new MqttProperties(); pubProps.setUserProperties(List.of(new UserProperty("k1", "v1"))); msg.setProperties(pubProps); pub.publish(pubTopic, msg).waitForCompletion(10_000); // 3. RECEIVE. if (!received.await(10, TimeUnit.SECONDS)) { throw new IllegalStateException("timed out waiting for the event"); } System.out.printf("received: %s%n", body[0]); pub.disconnect().waitForCompletion(5_000); pub.close(); sub.disconnect().waitForCompletion(5_000); sub.close(); } } ``` ```typescript import mqtt, { type MqttClient } from "mqtt"; async function main(): Promise { const url = process.env["KUBEMQ_MQTT_URL"] ?? "tcp://localhost:1883"; // '+' single-level wildcard -> KubeMQ '*'; '/' -> '.' in the channel name. const subFilter = "events/demo/+"; // channel filter demo.* const pubTopic = "events/demo/x"; // channel demo.x // 1. SUBSCRIBE FIRST — Events have no replay. const subscriber: MqttClient = mqtt.connect(url, { clientId: "js-events-sub", protocolVersion: 5, clean: true, keepalive: 30, }); await new Promise((res, rej) => { subscriber.once("connect", () => res()); subscriber.once("error", rej); }); const received = new Promise((resolve) => { subscriber.on("message", (_topic, payload) => resolve(payload.toString())); }); await new Promise((resolve, reject) => { subscriber.subscribe(subFilter, { qos: 1 }, (err, granted) => { if (err) return reject(err); // A granted qos >= 0x80 is a rejection (0xA2 = wildcard on a non-events pattern). for (const g of granted ?? []) { if ((g as { qos: number }).qos >= 0x80) return reject(new Error("subscription rejected")); } resolve(); }); }); await new Promise((r) => setTimeout(r, 300)); // let the subscription register // 2. PUBLISH with a v5 User Property (round-trips as a KubeMQ Tag). const publisher: MqttClient = mqtt.connect(url, { clientId: "js-events-pub", protocolVersion: 5, clean: true, keepalive: 30, }); await new Promise((res, rej) => { publisher.once("connect", () => res()); publisher.once("error", rej); }); await new Promise((resolve, reject) => { publisher.publish( pubTopic, "hello", { qos: 1, properties: { userProperties: { k1: "v1" } } }, // never set retain (err) => (err ? reject(err) : resolve()), ); }); // 3. RECEIVE. console.log("received:", await received); 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(); // '+' single-level wildcard -> KubeMQ '*'; '/' -> '.' in the channel name. const string subFilter = "events/demo/+"; // channel filter demo.* const string pubTopic = "events/demo/x"; // channel demo.x var factory = new MqttFactory(); var received = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); // 1. SUBSCRIBE FIRST — Events have no replay. using var sub = factory.CreateMqttClient(); sub.ApplicationMessageReceivedAsync += e => { received.TrySetResult(Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment)); return Task.CompletedTask; }; var subOpts = new MqttClientOptionsBuilder() .WithTcpServer(host, port) .WithClientId($"csharp-events-sub-{Guid.NewGuid():N}"[..26]) .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500) .WithCleanSession(true) .Build(); await sub.ConnectAsync(subOpts); var subResult = await sub.SubscribeAsync(new MqttClientSubscribeOptionsBuilder() .WithTopicFilter(f => f.WithTopic(subFilter).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)) .Build()); // A result code beyond GrantedQoS2 is a rejection (0xA2 = wildcard on a non-events pattern). if (subResult.Items.First().ResultCode > MqttClientSubscribeResultCode.GrantedQoS2) throw new Exception("subscription rejected"); await Task.Delay(300); // let the subscription register // 2. PUBLISH with a v5 User Property (round-trips as a KubeMQ Tag). using var pub = factory.CreateMqttClient(); var pubOpts = new MqttClientOptionsBuilder() .WithTcpServer(host, port) .WithClientId($"csharp-events-pub-{Guid.NewGuid():N}"[..26]) .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500) .WithCleanSession(true) .Build(); await pub.ConnectAsync(pubOpts); var pubResult = await pub.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic(pubTopic) .WithPayload(Encoding.UTF8.GetBytes("hello")) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithUserProperty("k1", "v1") // round-trips as a KubeMQ Tag; never set retain .Build()); if (pubResult.ReasonCode != MqttClientPublishReasonCode.Success) throw new Exception($"PUBACK reason: {pubResult.ReasonCode}"); // 3. RECEIVE. using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10)); Console.WriteLine($"received: {await received.Task.WaitAsync(cts.Token)}"); await pub.DisconnectAsync(); await sub.DisconnectAsync(); ``` ```ruby # The mqtt gem speaks MQTT 3.1.1 — no User Properties. Publish/subscribe on # Events works on any version; only the v5 Tag round-trip is unavailable here. 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" } # '+' single-level wildcard -> KubeMQ '*'; '/' -> '.' in the channel name. sub_filter = "events/demo/+" # channel filter demo.* pub_topic = "events/demo/x" # channel demo.x received = Queue.new sub_ready = Queue.new # 1. SUBSCRIBE FIRST — Events have no replay. subscriber = Thread.new do MQTT::Client.connect(host: conn[:host], port: conn[:port], ssl: conn[:ssl], client_id: "ruby-events-sub-#{SecureRandom.hex(4)}", clean_session: true, keep_alive: 30) do |client| client.subscribe([sub_filter, 1]) sub_ready.push(:ready) _topic, payload = client.get received.push(payload) end end sub_ready.pop # wait until the subscription is established # 2. PUBLISH (retain MUST be false — the broker silently drops retained messages). MQTT::Client.connect(host: conn[:host], port: conn[:port], ssl: conn[:ssl], client_id: "ruby-events-pub-#{SecureRandom.hex(4)}", clean_session: true, keep_alive: 30) do |client| client.publish(pub_topic, "hello", false, 1) end # 3. RECEIVE. payload = Timeout.timeout(10) { received.pop } puts "received: #{payload.inspect}" subscriber.kill ``` ```rust use rumqttc::v5::mqttbytes::v5::{Packet, PublishProperties}; 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; 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-level wildcard -> KubeMQ '*'; '/' -> '.' in the channel name. let sub_filter = "events/demo/+"; // channel filter demo.* let pub_topic = "events/demo/x"; // channel demo.x // 1. SUBSCRIBE FIRST — Events have no replay. let mut sub_opts = MqttOptions::new(format!("rust-events-sub-{sfx}"), &host, port); sub_opts.set_keep_alive(Duration::from_secs(30)); let (sub, mut sub_loop) = AsyncClient::new(sub_opts, 10); let (ready_tx, ready_rx) = oneshot::channel::<()>(); let (msg_tx, msg_rx) = oneshot::channel::(); tokio::spawn(async move { let mut ready = Some(ready_tx); let mut msg = Some(msg_tx); loop { match sub_loop.poll().await { Ok(Event::Incoming(Packet::SubAck(_))) => { if let Some(tx) = ready.take() { let _ = tx.send(()); } } 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; } } } }); sub.subscribe(sub_filter, QoS::AtLeastOnce).await?; timeout(Duration::from_secs(5), ready_rx).await??; // SUBACK received // 2. PUBLISH with a v5 User Property (round-trips as a KubeMQ Tag). let mut pub_opts = MqttOptions::new(format!("rust-events-pub-{sfx}"), &host, port); pub_opts.set_keep_alive(Duration::from_secs(30)); let (publisher, mut pub_loop) = AsyncClient::new(pub_opts, 10); tokio::spawn(async move { while pub_loop.poll().await.is_ok() {} }); let props = PublishProperties { user_properties: vec![("k1".to_string(), "v1".to_string())], ..Default::default() }; publisher .publish_with_properties(pub_topic, QoS::AtLeastOnce, false, b"hello".as_ref(), props) .await?; // 3. RECEIVE. let body = timeout(Duration::from_secs(10), msg_rx).await??; println!("received: {body}"); Ok(()) } ``` ## Wildcard subscriptions [#wildcard-subscriptions] Wildcard filters are accepted on the **Events pattern only**. Two wildcards map onto the broker's native channel wildcards: | MQTT wildcard | KubeMQ wildcard | Matches | | ------------- | --------------- | ---------------------------------------------------------------------------------------------------- | | `+` | `*` | exactly one segment — `events/demo/+` matches `events/demo/x`, not `events/demo/a/b` | | `#` | `>` | one or more trailing segments (must be the last token) — `events/demo/#` matches `events/demo/a/b/c` | A bare `#` subscribes to all Events topics (it resolves through the `events` `DefaultPattern`). A wildcard on any non-Events prefix (`store/#`, `queues/#`, …) is rejected with SUBACK `0xA2`. **One filter per subscription.** Each distinct subscribe filter is an independent bridge entry. If several of your active filters match the same publish, you receive **one copy per matching filter** — there is no cross-filter de-duplication. Use a single, specific filter per subscription to avoid duplicate delivery. ## User Properties and Tags (MQTT 5.0) [#user-properties-and-tags-mqtt-50] On MQTT 5.0 connections, a publisher's `PUBLISH` User Properties are copied to the KubeMQ message `Tags`, and a v5 subscriber receives those `Tags` back as User Properties on delivery. Duplicate keys: last wins. The caps are **32 properties** and **4096 bytes total** (all key + value lengths); exceeding either returns PUBACK `0x97` and the message is dropped. MQTT 3.1.1 connections carry no User Properties — nothing is propagated in either direction. ## Related [#related]