Events

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})
}

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()

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();
    }
}

import mqtt, { type MqttClient } from "mqtt";

async function main(): Promise<void> {
  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<void>((res, rej) => {
    subscriber.once("connect", () => res());
    subscriber.once("error", rej);
  });

  const received = new Promise<string>((resolve) => {
    subscriber.on("message", (_topic, payload) => resolve(payload.toString()));
  });
  await new Promise<void>((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<void>((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<void>((res, rej) => {
    publisher.once("connect", () => res());
    publisher.once("error", rej);
  });
  await new Promise<void>((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);
});

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<string>(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();

# 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

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<dyn std::error::Error>> {
    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::<String>();
    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(())
}