Getting Started

package main

import (
	"context"
	"fmt"
	"log"
	"net"
	"os"
	"strings"
	"time"

	"github.com/eclipse/paho.golang/paho"
)

func brokerURL() string {
	if u := os.Getenv("KUBEMQ_MQTT_URL"); u != "" {
		return u
	}
	return "tcp://localhost:1883"
}

func tcpAddr(rawURL string) string {
	for _, pfx := range []string{"tcp://", "ws://", "tls://"} {
		if strings.HasPrefix(rawURL, pfx) {
			return rawURL[len(pfx):]
		}
	}
	return rawURL
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()

	// Subscribe with the single-level '+' wildcard (KubeMQ '*'); publish to a
	// concrete leaf it matches. "events/demo/x" maps to channel "demo.x".
	const subFilter = "events/demo/+"
	const pubTopic = "events/demo/x"
	addr := tcpAddr(brokerURL())
	recvCh := make(chan string, 1)

	// SUBSCRIBER — connect and subscribe before publishing.
	subConn, err := net.Dial("tcp", addr)
	if err != nil {
		log.Fatalf("sub dial: %v", err)
	}
	subClient := paho.NewClient(paho.ClientConfig{
		Conn: subConn,
		OnPublishReceived: []func(paho.PublishReceived) (bool, error){
			func(pr paho.PublishReceived) (bool, error) {
				recvCh <- string(pr.Packet.Payload)
				return true, nil
			},
		},
	})
	if _, err := subClient.Connect(ctx, &paho.Connect{ClientID: "go-sub", KeepAlive: 30, CleanStart: true}); err != nil {
		log.Fatalf("sub connect: %v", err)
	}
	subAck, err := subClient.Subscribe(ctx, &paho.Subscribe{
		Subscriptions: []paho.SubscribeOptions{{Topic: subFilter, QoS: 1}},
	})
	if err != nil {
		log.Fatalf("subscribe: %v", err)
	}
	// Reason code > 2 is a rejection (0xA2 wildcard-not-supported, 0x83 impl-specific).
	if subAck.Reasons[0] > 2 {
		log.Fatalf("SUBACK reason=0x%02X", subAck.Reasons[0])
	}
	fmt.Printf("[sub] subscribed to %q\n", subFilter)
	time.Sleep(300 * time.Millisecond) // let the subscription register

	// PUBLISHER — connect and publish one QoS-1 message.
	pubConn, err := net.Dial("tcp", addr)
	if err != nil {
		log.Fatalf("pub dial: %v", err)
	}
	pubClient := paho.NewClient(paho.ClientConfig{Conn: pubConn})
	if _, err := pubClient.Connect(ctx, &paho.Connect{ClientID: "go-pub", KeepAlive: 30, CleanStart: true}); err != nil {
		log.Fatalf("pub connect: %v", err)
	}
	if _, err := pubClient.Publish(ctx, &paho.Publish{Topic: pubTopic, QoS: 1, Payload: []byte("hello")}); err != nil {
		log.Fatalf("publish: %v", err)
	}
	fmt.Printf("[pub] published to %q\n", pubTopic)

	// RECEIVE — wait for the subscriber to get the message.
	select {
	case payload := <-recvCh:
		fmt.Printf("[sub] received: %s (channel demo.x)\n", payload)
	case <-ctx.Done():
		log.Fatal("timed out waiting for the event")
	}

	_ = subClient.Disconnect(&paho.Disconnect{ReasonCode: 0})
	_ = pubClient.Disconnect(&paho.Disconnect{ReasonCode: 0})
}

import os
import threading
import time

import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion


def parse_url(url: str) -> tuple[str, int]:
    scheme, rest = url.split("://", 1)
    host_port = rest.rstrip("/")
    if ":" in host_port:
        host, port = host_port.rsplit(":", 1)
        return host, int(port)
    return host_port, {"tcp": 1883, "tls": 8883, "ws": 8083}.get(scheme, 1883)


def main() -> None:
    host, port = parse_url(os.environ.get("KUBEMQ_MQTT_URL", "tcp://localhost:1883"))

    # Subscribe with the single-level "+" wildcard; publish to a matching leaf.
    sub_topic = "events/demo/+"  # KubeMQ channel demo.*
    pub_topic = "events/demo/x"  # KubeMQ channel demo.x

    received = threading.Event()

    # SUBSCRIBER
    sub = mqtt.Client(callback_api_version=CallbackAPIVersion.VERSION2,
                      client_id="python-sub", protocol=mqtt.MQTTv5)

    def on_connect(client, userdata, flags, rc, props):
        client.subscribe(sub_topic, qos=1)

    def on_message(client, userdata, msg):
        print(f"[sub] received: {msg.payload.decode()!r} (channel demo.x)")
        received.set()

    sub.on_connect = on_connect
    sub.on_message = on_message
    sub.connect(host, port, keepalive=30, clean_start=True)
    sub.loop_start()
    time.sleep(0.5)  # let the subscription register

    # PUBLISHER
    pub = mqtt.Client(callback_api_version=CallbackAPIVersion.VERSION2,
                      client_id="python-pub", protocol=mqtt.MQTTv5)
    pub.connect(host, port, keepalive=30, clean_start=True)
    pub.loop_start()
    pub.publish(pub_topic, payload=b"hello", qos=1).wait_for_publish(timeout=10)
    print(f"[pub] published to {pub_topic!r}")

    if not received.wait(timeout=10):
        raise TimeoutError("timed out waiting for the event")

    pub.loop_stop(); pub.disconnect()
    sub.loop_stop(); sub.disconnect()


if __name__ == "__main__":
    main()

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.client.persist.MemoryPersistence;
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 {
    public static void main(String[] args) throws Exception {
        String broker = System.getenv().getOrDefault("KUBEMQ_MQTT_URL", "tcp://localhost:1883");
        String subTopic = "events/demo/+";  // single-level '+' wildcard
        String pubTopic = "events/demo/x";  // KubeMQ channel demo.x

        MqttConnectionOptions opts = new MqttConnectionOptions();
        opts.setCleanStart(true);
        opts.setKeepAliveInterval(30);

        CountDownLatch received = new CountDownLatch(1);

        // SUBSCRIBER
        MqttAsyncClient sub = new MqttAsyncClient(broker,
                "java-sub-" + UUID.randomUUID().toString().substring(0, 8), new MemoryPersistence());
        sub.setCallback(new MqttCallback() {
            @Override public void messageArrived(String topic, MqttMessage msg) {
                System.out.printf("[sub] received: %s (channel demo.x)%n",
                        new String(msg.getPayload(), StandardCharsets.UTF_8));
                received.countDown();
            }
            @Override public void disconnected(MqttDisconnectResponse r) { }
            @Override public void mqttErrorOccurred(MqttException e) { }
            @Override public void deliveryComplete(IMqttToken t) { }
            @Override public void connectComplete(boolean reconnect, String uri) { }
            @Override public void authPacketArrived(int code, MqttProperties p) { }
        });
        sub.connect(opts).waitForCompletion(10_000);
        sub.subscribe(subTopic, 1).waitForCompletion(10_000);
        System.out.printf("[sub] subscribed to '%s'%n", subTopic);

        // PUBLISHER
        MqttAsyncClient pub = new MqttAsyncClient(broker,
                "java-pub-" + UUID.randomUUID().toString().substring(0, 8), new MemoryPersistence());
        pub.connect(opts).waitForCompletion(10_000);
        MqttMessage msg = new MqttMessage("hello".getBytes(StandardCharsets.UTF_8));
        msg.setQos(1);
        pub.publish(pubTopic, msg).waitForCompletion(10_000);
        System.out.printf("[pub] published to '%s'%n", pubTopic);

        if (!received.await(10, TimeUnit.SECONDS)) {
            throw new IllegalStateException("timed out waiting for the event");
        }

        pub.disconnect().waitForCompletion(5_000); pub.close();
        sub.disconnect().waitForCompletion(5_000); sub.close();
    }
}

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

function brokerUrl(): string {
  return process.env["KUBEMQ_MQTT_URL"] ?? "tcp://localhost:1883";
}

async function main(): Promise<void> {
  const url = brokerUrl();
  const subFilter = "events/demo/+"; // single-level '+' wildcard
  const pubTopic = "events/demo/x"; // KubeMQ channel demo.x

  // SUBSCRIBER
  const subscriber: MqttClient = mqtt.connect(url, {
    clientId: "js-sub",
    protocolVersion: 5,
    clean: true,
  });
  await new Promise<void>((resolve, reject) => {
    subscriber.once("connect", () => resolve());
    subscriber.once("error", reject);
  });

  const received = new Promise<void>((resolve) => {
    subscriber.on("message", (topic, payload) => {
      console.log(`[sub] received: ${payload.toString()} (channel demo.x)`);
      resolve();
    });
  });
  await new Promise<void>((resolve, reject) => {
    subscriber.subscribe(subFilter, { qos: 1 }, (err) => (err ? reject(err) : resolve()));
  });
  console.log(`[sub] subscribed to ${subFilter}`);
  await new Promise<void>((r) => setTimeout(r, 300));

  // PUBLISHER
  const publisher: MqttClient = mqtt.connect(url, {
    clientId: "js-pub",
    protocolVersion: 5,
    clean: true,
  });
  await new Promise<void>((resolve, reject) => {
    publisher.once("connect", () => resolve());
    publisher.once("error", reject);
  });
  await new Promise<void>((resolve, reject) => {
    publisher.publish(pubTopic, "hello", { qos: 1, retain: false }, (err) =>
      err ? reject(err) : resolve());
  });
  console.log(`[pub] published to ${pubTopic}`);

  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) ParseEndpoint(string url)
{
    foreach (var prefix in new[] { "tcp://", "tls://", "ws://" })
        if (url.StartsWith(prefix, StringComparison.OrdinalIgnoreCase))
            url = url[prefix.Length..];
    var parts = url.TrimEnd('/').Split(':');
    return (parts[0], parts.Length > 1 ? int.Parse(parts[1]) : 1883);
}

var url = Environment.GetEnvironmentVariable("KUBEMQ_MQTT_URL") ?? "tcp://localhost:1883";
var (host, port) = ParseEndpoint(url);

const string subTopic = "events/demo/+"; // single-level '+' wildcard
const string pubTopic = "events/demo/x"; // KubeMQ channel demo.x

var factory = new MqttFactory();
var received = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);

// SUBSCRIBER
using var subClient = factory.CreateMqttClient();
subClient.ApplicationMessageReceivedAsync += e =>
{
    received.TrySetResult(Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment));
    return Task.CompletedTask;
};
var subOptions = new MqttClientOptionsBuilder()
    .WithTcpServer(host, port)
    .WithClientId("csharp-sub")
    .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
    .WithCleanSession(true)
    .Build();
await subClient.ConnectAsync(subOptions);
await subClient.SubscribeAsync(new MqttClientSubscribeOptionsBuilder()
    .WithTopicFilter(f => f.WithTopic(subTopic).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
    .Build());
Console.WriteLine($"[sub] subscribed to '{subTopic}'");
await Task.Delay(300);

// PUBLISHER
using var pubClient = factory.CreateMqttClient();
var pubOptions = new MqttClientOptionsBuilder()
    .WithTcpServer(host, port)
    .WithClientId("csharp-pub")
    .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
    .WithCleanSession(true)
    .Build();
await pubClient.ConnectAsync(pubOptions);
await pubClient.PublishAsync(new MqttApplicationMessageBuilder()
    .WithTopic(pubTopic)
    .WithPayload(Encoding.UTF8.GetBytes("hello"))
    .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
    .Build());
Console.WriteLine($"[pub] published to '{pubTopic}'");

using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
var body = await received.Task.WaitAsync(cts.Token);
Console.WriteLine($"[sub] received: {body} (channel demo.x)");

await subClient.DisconnectAsync();
await pubClient.DisconnectAsync();

# The Ruby `mqtt` gem speaks MQTT 3.1.1 only.
require "mqtt"
require "uri"
require "timeout"

uri = URI.parse(ENV.fetch("KUBEMQ_MQTT_URL", "tcp://localhost:1883"))
conn = { host: uri.host, port: uri.port, ssl: uri.scheme == "tls" }

sub_topic = "events/demo/+" # single-level '+' wildcard
pub_topic = "events/demo/x" # KubeMQ channel demo.x

received = Queue.new
sub_ready = Queue.new

# SUBSCRIBER thread
subscriber = Thread.new do
  MQTT::Client.connect(**conn, client_id: "ruby-sub", clean_session: true) do |client|
    client.subscribe([sub_topic, 1])
    sub_ready.push(:ready)
    topic, payload = client.get
    received.push(payload)
  end
end

sub_ready.pop
puts "[sub] subscribed to '#{sub_topic}'"

# PUBLISHER
MQTT::Client.connect(**conn, client_id: "ruby-pub", clean_session: true) do |client|
  # retain=false is mandatory — the broker silently drops retained publishes.
  client.publish(pub_topic, "hello", false, 1)
  puts "[pub] published to '#{pub_topic}'"
end

payload = Timeout.timeout(10) { received.pop }
puts "[sub] received: #{payload.inspect} (channel demo.x)"
subscriber.kill

use rumqttc::v5::mqttbytes::v5::Packet;
use rumqttc::v5::mqttbytes::QoS;
use rumqttc::v5::{AsyncClient, Event, MqttOptions};
use std::env;
use std::time::Duration;
use tokio::sync::oneshot;
use tokio::time::timeout;

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 sub_topic = "events/demo/+"; // single-level '+' wildcard
    let pub_topic = "events/demo/x"; // KubeMQ channel demo.x

    // SUBSCRIBER
    let mut sub_opts = MqttOptions::new("rust-sub", &host, port);
    sub_opts.set_keep_alive(Duration::from_secs(30));
    let (sub_client, 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(_) => return,
            }
        }
    });
    sub_client.subscribe(sub_topic, QoS::AtLeastOnce).await?;
    timeout(Duration::from_secs(5), ready_rx).await??;
    println!("[sub] subscribed to '{sub_topic}'");

    // PUBLISHER
    let mut pub_opts = MqttOptions::new("rust-pub", &host, port);
    pub_opts.set_keep_alive(Duration::from_secs(30));
    let (pub_client, mut pub_loop) = AsyncClient::new(pub_opts, 10);
    tokio::spawn(async move { while pub_loop.poll().await.is_ok() {} });

    // retain = false — a retained publish is silently dropped.
    pub_client.publish(pub_topic, QoS::AtLeastOnce, false, b"hello".as_ref()).await?;
    println!("[pub] published to '{pub_topic}'");

    let payload = timeout(Duration::from_secs(10), msg_rx).await??;
    println!("[sub] received: {payload} (channel demo.x)");
    Ok(())
}