MQTT

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

// tcpAddr strips the scheme prefix to obtain "host:port".
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()

	// Prefix "events/" selects the Events pattern; '/' becomes '.' so
	// "events/demo/x" maps to the KubeMQ channel "demo.x".
	const topic = "events/demo/x"

	conn, err := net.Dial("tcp", tcpAddr(brokerURL()))
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	client := paho.NewClient(paho.ClientConfig{Conn: conn})

	connAck, err := client.Connect(ctx, &paho.Connect{
		ClientID:   "go-mqtt-events-pub",
		KeepAlive:  30,
		CleanStart: true,
	})
	if err != nil {
		log.Fatalf("connect: %v", err)
	}
	if connAck.ReasonCode != 0 {
		log.Fatalf("CONNACK reason=0x%02X", connAck.ReasonCode)
	}

	// QoS 1 returns a PUBACK. Do NOT set Retain — a retained publish is
	// silently dropped (PUBACK 0x00, message never delivered).
	pubAck, err := client.Publish(ctx, &paho.Publish{
		Topic:   topic,
		QoS:     1,
		Payload: []byte("hello from MQTT"),
		Properties: &paho.PublishProperties{
			// MQTT 5.0 User Properties round-trip as KubeMQ Tags.
			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)
	}
	fmt.Printf("published to %q (channel demo.x)\n", topic)

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

import os

import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes


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:
    url = os.environ.get("KUBEMQ_MQTT_URL", "tcp://localhost:1883")
    host, port = parse_url(url)

    # Prefix "events/" selects the Events pattern; "/" becomes "." so
    # "events/demo/x" maps to the KubeMQ channel "demo.x".
    topic = "events/demo/x"

    # paho-mqtt 2.x requires an explicit callback API version.
    client = mqtt.Client(
        callback_api_version=CallbackAPIVersion.VERSION2,
        client_id="python-mqtt-events-pub",
        protocol=mqtt.MQTTv5,
    )
    client.connect(host, port, keepalive=30, clean_start=True)
    client.loop_start()

    # MQTT 5.0 User Properties round-trip as KubeMQ Tags.
    props = Properties(PacketTypes.PUBLISH)
    props.UserProperty = [("sensor", "thermometer")]

    # Do not set retain=True — a retained publish is silently dropped.
    info = client.publish(topic, payload=b"hello from MQTT", qos=1, properties=props)
    info.wait_for_publish(timeout=10)
    print(f"published to {topic!r} (channel demo.x)")

    client.loop_stop()
    client.disconnect()


if __name__ == "__main__":
    main()

import java.nio.charset.StandardCharsets;
import java.util.List;

import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
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");

        // Prefix "events/" selects the Events pattern; "/" becomes "." so
        // "events/demo/x" maps to the KubeMQ channel "demo.x".
        String topic = "events/demo/x";

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

        // org.eclipse.paho.mqttv5.client always uses MQTT protocol level 5.
        MqttAsyncClient client = new MqttAsyncClient(broker, "java-mqtt-events-pub", new MemoryPersistence());
        client.connect(opts).waitForCompletion(10_000);

        MqttMessage msg = new MqttMessage("hello from MQTT".getBytes(StandardCharsets.UTF_8));
        msg.setQos(1);
        // Do NOT call msg.setRetained(true) — a retained publish is silently dropped.

        // MQTT 5.0 User Properties round-trip as KubeMQ Tags.
        MqttProperties props = new MqttProperties();
        props.setUserProperties(List.of(new UserProperty("sensor", "thermometer")));
        msg.setProperties(props);

        client.publish(topic, msg).waitForCompletion(10_000);
        System.out.printf("published to %s (channel demo.x)%n", topic);

        client.disconnect().waitForCompletion(5_000);
        client.close();
    }
}

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

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

async function main(): Promise<void> {
  // Prefix "events/" selects the Events pattern; "/" becomes "." so
  // "events/demo/x" maps to the KubeMQ channel "demo.x".
  const topic = "events/demo/x";

  const client: MqttClient = mqtt.connect(brokerUrl(), {
    clientId: "js-mqtt-events-pub",
    protocolVersion: 5,
    clean: true,
    keepalive: 30,
  });

  await new Promise<void>((resolve, reject) => {
    client.once("connect", () => resolve());
    client.once("error", reject);
  });

  await new Promise<void>((resolve, reject) => {
    client.publish(
      topic,
      "hello from MQTT",
      {
        qos: 1,
        retain: false, // retain is NOT supported; a retained publish is silently dropped.
        // MQTT 5.0 User Properties round-trip as KubeMQ Tags.
        properties: { userProperties: { sensor: "thermometer" } },
      },
      (err) => (err ? reject(err) : resolve()),
    );
  });
  console.log(`published to "${topic}" (channel demo.x)`);

  await client.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);

// Prefix "events/" selects the Events pattern; "/" becomes "." so
// "events/demo/x" maps to the KubeMQ channel "demo.x".
const string topic = "events/demo/x";

var factory = new MqttFactory();
using var client = factory.CreateMqttClient();

var options = new MqttClientOptionsBuilder()
    .WithTcpServer(host, port)
    .WithClientId("csharp-mqtt-events-pub")
    .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
    .WithCleanSession(true)
    .Build();

await client.ConnectAsync(options);

var pubResult = await client.PublishAsync(
    new MqttApplicationMessageBuilder()
        .WithTopic(topic)
        .WithPayload(Encoding.UTF8.GetBytes("hello from MQTT"))
        .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
        // MQTT 5.0 User Property round-trips as a KubeMQ Tag.
        // Do NOT call WithRetainFlag(true) — a retained publish is silently dropped.
        .WithUserProperty("sensor", "thermometer")
        .Build());

if (pubResult.ReasonCode != MqttClientPublishReasonCode.Success)
    throw new Exception($"PUBACK reason: {pubResult.ReasonCode}");
Console.WriteLine($"published to '{topic}' (channel demo.x)");

await client.DisconnectAsync();

# The Ruby `mqtt` gem speaks MQTT 3.1.1 only (no User Properties, no v5 RPC).
require "mqtt"
require "uri"

uri = URI.parse(ENV.fetch("KUBEMQ_MQTT_URL", "tcp://localhost:1883"))

# Prefix "events/" selects the Events pattern; "/" becomes "." so
# "events/demo/x" maps to the KubeMQ channel "demo.x".
topic = "events/demo/x"

MQTT::Client.connect(
  host:          uri.host,
  port:          uri.port,
  ssl:           uri.scheme == "tls",
  client_id:     "ruby-mqtt-events-pub",
  clean_session: true,
  keep_alive:    30
) do |client|
  # retain=false is mandatory — the broker silently drops retained publishes.
  client.publish(topic, "hello from MQTT", false, 1)
  puts "published to '#{topic}' (channel demo.x)"
end

use rumqttc::v5::mqttbytes::v5::PublishProperties;
use rumqttc::v5::mqttbytes::QoS;
use rumqttc::v5::{AsyncClient, MqttOptions};
use std::env;
use std::time::Duration;

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

    // Prefix "events/" selects the Events pattern; '/' becomes '.' so
    // "events/demo/x" maps to the KubeMQ channel "demo.x".
    let topic = "events/demo/x";

    let mut opts = MqttOptions::new("rust-mqtt-events-pub", host, port);
    opts.set_keep_alive(Duration::from_secs(30));
    let (client, mut eventloop) = AsyncClient::new(opts, 10);

    // Drive the event loop so the PUBLISH is flushed and the PUBACK is processed.
    tokio::spawn(async move { while eventloop.poll().await.is_ok() {} });

    // MQTT 5.0 User Properties round-trip as KubeMQ Tags.
    let props = PublishProperties {
        user_properties: vec![("sensor".to_string(), "thermometer".to_string())],
        ..Default::default()
    };

    // retain = false — a retained publish is silently dropped.
    client
        .publish_with_properties(topic, QoS::AtLeastOnce, false, b"hello from MQTT".as_ref(), props)
        .await?;
    println!("published to '{topic}' (channel demo.x)");

    Ok(())
}