Queues

package main

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

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

const produceTopic = "queues/demo/q"                  // plain publish; channel demo.q
const consumeTopic = "$share/g1/queues/demo/q"        // MQTT 5.0 shared subscription

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(), 30*time.Second)
	defer cancel()
	addr := tcpAddr(brokerURL())
	sfx := uuid.NewString()[:8]

	// 1. CONSUME — MQTT 5.0 shared subscription. Returning true sends PUBACK,
	//    which is how KubeMQ marks the message acknowledged.
	got := make(chan string, 1)
	consumer := dial(ctx, addr, "go-q-worker-"+sfx, func(pr paho.PublishReceived) (bool, error) {
		got <- string(pr.Packet.Payload)
		return true, nil
	})
	subAck, err := consumer.Subscribe(ctx, &paho.Subscribe{
		Subscriptions: []paho.SubscribeOptions{{Topic: consumeTopic, QoS: 1}},
	})
	if err != nil {
		log.Fatalf("subscribe: %v", err)
	}
	// SUBACK 0x83 = plain queues/ subscribe, QoS-0 shared subscribe, or wrong pattern.
	if subAck.Reasons[0] == 0x83 {
		log.Fatal("SUBACK 0x83: use the $share/<group>/queues/<ch> form at QoS >= 1")
	}
	time.Sleep(300 * time.Millisecond)

	// 2. PRODUCE — plain publish to queues/<ch>.
	producer := dial(ctx, addr, "go-q-prod-"+sfx, nil)
	pubAck, err := producer.Publish(ctx, &paho.Publish{
		Topic: produceTopic, QoS: 1, Payload: []byte(`{"job":1}`),
	})
	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("consumed: %s (acked via PUBACK)\n", msg)
	case <-ctx.Done():
		log.Fatal("timed out waiting for the queue message")
	}
	_ = consumer.Disconnect(&paho.Disconnect{ReasonCode: 0})
	_ = producer.Disconnect(&paho.Disconnect{ReasonCode: 0})
}

import os
import threading
import time

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

PRODUCE_TOPIC = "queues/demo/q"            # plain publish; channel demo.q
CONSUME_TOPIC = "$share/g1/queues/demo/q"  # MQTT 5.0 shared subscription


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] = []
    subscribed = threading.Event()

    # 1. CONSUME — MQTT 5.0 shared subscription at QoS 1 (QoS 0 -> SUBACK 0x83).
    consumer = mqtt.Client(CallbackAPIVersion.VERSION2, client_id="py-q-consumer", protocol=mqtt.MQTTv5)
    consumer.on_connect = lambda c, *_: c.subscribe(CONSUME_TOPIC, qos=1)
    consumer.on_subscribe = lambda *_: subscribed.set()
    consumer.on_message = lambda c, u, m: (payload.append(m.payload), received.set())
    consumer.connect(host, port, keepalive=30)
    consumer.loop_start()
    if not subscribed.wait(timeout=10):
        raise TimeoutError("shared subscription not established")

    # 2. PRODUCE — plain publish to queues/<ch>.
    producer = mqtt.Client(CallbackAPIVersion.VERSION2, client_id="py-q-producer", protocol=mqtt.MQTTv5)
    producer.connect(host, port, keepalive=30)
    producer.loop_start()
    producer.publish(PRODUCE_TOPIC, payload=b'{"job": 1}', qos=1).wait_for_publish(10)

    # 3. RECEIVE. The auto-PUBACK acks the message to KubeMQ.
    if not received.wait(timeout=15):
        raise TimeoutError("timed out waiting for the queue message")
    print(f"consumed: {payload[0].decode()!r} (acked via PUBACK)")

    producer.loop_stop(); producer.disconnect()
    consumer.loop_stop(); consumer.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.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;

public final class Main {
    private static final String PRODUCE_TOPIC = "queues/demo/q";           // plain publish; channel demo.q
    private static final String CONSUME_TOPIC = "$share/g1/queues/demo/q"; // MQTT 5.0 shared subscription

    public static void main(String[] args) throws Exception {
        String broker = System.getenv().getOrDefault("KUBEMQ_MQTT_URL", "tcp://localhost:1883");
        MqttConnectionOptions opts = new MqttConnectionOptions();
        opts.setCleanStart(true);
        opts.setKeepAliveInterval(30);

        CountDownLatch received = new CountDownLatch(1);
        String[] body = {null};

        // 1. CONSUME — MQTT 5.0 shared subscription at QoS 1 (QoS 0 -> SUBACK 0x83).
        //    Paho auto-sends PUBACK when messageArrived returns, acking the message.
        MqttAsyncClient consumer = new MqttAsyncClient(broker, "java-q-consumer-" + UUID.randomUUID().toString().substring(0, 6));
        consumer.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) {}
        });
        consumer.connect(opts).waitForCompletion(10_000);
        IMqttToken subToken = consumer.subscribe(CONSUME_TOPIC, 1);
        subToken.waitForCompletion(10_000);
        // SUBACK 0x83 = plain queues/ subscribe, QoS-0 shared subscribe, or wrong pattern.
        if (subToken.getGrantedQos()[0] == 0x83) {
            throw new IllegalStateException("SUBACK 0x83: use $share/<group>/queues/<ch> at QoS >= 1");
        }
        Thread.sleep(400);

        // 2. PRODUCE — plain publish to queues/<ch>.
        MqttAsyncClient producer = new MqttAsyncClient(broker, "java-q-producer-" + UUID.randomUUID().toString().substring(0, 6));
        producer.connect(opts).waitForCompletion(10_000);
        MqttMessage msg = new MqttMessage("{\"job\":1}".getBytes(StandardCharsets.UTF_8));
        msg.setQos(1);
        producer.publish(PRODUCE_TOPIC, msg).waitForCompletion(5_000);

        // 3. RECEIVE.
        if (!received.await(15, TimeUnit.SECONDS)) {
            throw new IllegalStateException("timed out waiting for the queue message");
        }
        System.out.printf("consumed: %s (acked via PUBACK)%n", body[0]);

        producer.disconnect().waitForCompletion(5_000); producer.close();
        consumer.disconnect().waitForCompletion(5_000); consumer.close();
    }
}

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

const PRODUCE_TOPIC = "queues/demo/q";           // plain publish; channel demo.q
const CONSUME_TOPIC = "$share/g1/queues/demo/q"; // MQTT 5.0 shared subscription

function connect(url: string, clientId: string): Promise<MqttClient> {
  return new Promise((resolve, reject) => {
    const client = mqtt.connect(url, { clientId, protocolVersion: 5, clean: true, keepalive: 30 });
    client.once("connect", () => resolve(client));
    client.once("error", reject);
  });
}

async function main(): Promise<void> {
  const url = process.env["KUBEMQ_MQTT_URL"] ?? "tcp://localhost:1883";

  // 1. CONSUME — MQTT 5.0 shared subscription at QoS 1 (QoS 0 / plain queues/ -> SUBACK 0x83).
  const consumer = await connect(url, "js-q-consumer");
  const received = new Promise<string>((resolve) => {
    consumer.on("message", (_topic, payload) => resolve(payload.toString()));
  });
  await new Promise<void>((resolve, reject) => {
    consumer.subscribe(CONSUME_TOPIC, { qos: 1 }, (err) =>
      err ? reject(new Error("subscribe rejected — use $share/<group>/queues/<ch> at QoS >= 1")) : resolve(),
    );
  });
  await new Promise<void>((r) => setTimeout(r, 300));

  // 2. PRODUCE — plain publish to queues/<ch>.
  const producer = await connect(url, "js-q-producer");
  await new Promise<void>((resolve, reject) => {
    producer.publish(PRODUCE_TOPIC, JSON.stringify({ job: 1 }), { qos: 1 }, (err) =>
      err ? reject(err) : resolve(),
    );
  });

  // 3. RECEIVE. The auto-PUBACK acks the message to KubeMQ.
  console.log(`consumed: ${await received} (acked via PUBACK)`);

  await producer.endAsync();
  await consumer.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();
const string produceTopic = "queues/demo/q";           // plain publish; channel demo.q
const string consumeTopic = "$share/g1/queues/demo/q"; // MQTT 5.0 shared subscription

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

MqttClientOptions Options(string id) => new MqttClientOptionsBuilder()
    .WithTcpServer(host, port)
    .WithClientId($"{id}-{Guid.NewGuid():N}"[..26])
    .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
    .WithCleanSession(true)
    .Build();

// 1. CONSUME — MQTT 5.0 shared subscription at QoS 1 (QoS 0 / plain queues/ -> SUBACK 0x83).
using var consumer = factory.CreateMqttClient();
consumer.ApplicationMessageReceivedAsync += e =>
{
    received.TrySetResult(Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment));
    return Task.CompletedTask;
};
await consumer.ConnectAsync(Options("csharp-q-consumer"));
var subResult = await consumer.SubscribeAsync(new MqttClientSubscribeOptionsBuilder()
    .WithTopicFilter(f => f.WithTopic(consumeTopic).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
    .Build());
if (subResult.Items.First().ResultCode > MqttClientSubscribeResultCode.GrantedQoS2)
    throw new Exception("subscribe rejected — use $share/<group>/queues/<ch> at QoS >= 1");
await Task.Delay(300);

// 2. PRODUCE — plain publish to queues/<ch>.
using var producer = factory.CreateMqttClient();
await producer.ConnectAsync(Options("csharp-q-producer"));
var pubResult = await producer.PublishAsync(new MqttApplicationMessageBuilder()
    .WithTopic(produceTopic)
    .WithPayload(Encoding.UTF8.GetBytes("{\"job\":1}"))
    .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
    .Build());
if (pubResult.ReasonCode != MqttClientPublishReasonCode.Success)
    throw new Exception($"PUBACK reason: {pubResult.ReasonCode}");

// 3. RECEIVE. The auto-PUBACK acks the message to KubeMQ.
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
Console.WriteLine($"consumed: {await received.Task.WaitAsync(cts.Token)} (acked via PUBACK)");

await producer.DisconnectAsync();
await consumer.DisconnectAsync();

use rumqttc::v5::mqttbytes::v5::Packet;
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;

const PRODUCE_TOPIC: &str = "queues/demo/q";           // plain publish; channel demo.q
const CONSUME_TOPIC: &str = "$share/g1/queues/demo/q"; // MQTT 5.0 shared subscription

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

    // 1. CONSUME — MQTT 5.0 shared subscription. rumqttc auto-sends PUBACK for
    //    QoS 1, which acks the message and removes it from the queue.
    let mut con_opts = MqttOptions::new(format!("rust-q-con-{sfx}"), &host, port);
    con_opts.set_keep_alive(Duration::from_secs(30));
    let (consumer, mut con_loop) = AsyncClient::new(con_opts, 10);

    let (msg_tx, msg_rx) = oneshot::channel::<String>();
    tokio::spawn(async move {
        let mut msg = Some(msg_tx);
        loop {
            match con_loop.poll().await {
                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; }
            }
        }
    });
    consumer.subscribe(CONSUME_TOPIC, QoS::AtLeastOnce).await?;
    tokio::time::sleep(Duration::from_millis(400)).await; // SUBACK round-trip

    // 2. PRODUCE — plain publish to queues/<ch>.
    let mut prod_opts = MqttOptions::new(format!("rust-q-prod-{sfx}"), &host, port);
    prod_opts.set_keep_alive(Duration::from_secs(30));
    let (producer, mut prod_loop) = AsyncClient::new(prod_opts, 10);
    tokio::spawn(async move { while prod_loop.poll().await.is_ok() {} });
    producer.publish(PRODUCE_TOPIC, QoS::AtLeastOnce, false, br#"{"job":1}"#.as_ref()).await?;

    // 3. RECEIVE.
    let body = timeout(Duration::from_secs(15), msg_rx).await??;
    println!("consumed: {body} (acked via PUBACK)");
    Ok(())
}