Commands

package main

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

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

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 main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	addr := tcpAddr(brokerURL())

	clientID := "go-cmd-" + uuid.NewString()[:8]
	correlationID := uuid.NewString()
	replyTopic := "$reply/" + clientID + "/inbox" // own namespace only
	commandTopic := "commands/demo/cmd"            // channel demo.cmd

	resp := make(chan *paho.Publish, 1)
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		log.Fatalf("dial: %v", err)
	}
	client := paho.NewClient(paho.ClientConfig{
		Conn: conn,
		OnPublishReceived: []func(paho.PublishReceived) (bool, error){
			func(pr paho.PublishReceived) (bool, error) {
				select {
				case resp <- pr.Packet:
				default:
				}
				return true, nil
			},
		},
	})
	ack, err := client.Connect(ctx, &paho.Connect{ClientID: clientID, KeepAlive: 30, CleanStart: true})
	if err != nil || ack.ReasonCode != 0 {
		log.Fatalf("connect: %v (reason 0x%02X)", err, ack.ReasonCode)
	}

	// 1. Subscribe to your own reply topic BEFORE publishing.
	subAck, err := client.Subscribe(ctx, &paho.Subscribe{
		Subscriptions: []paho.SubscribeOptions{{Topic: replyTopic, QoS: 1}},
	})
	if err != nil {
		log.Fatalf("subscribe: %v", err)
	}
	// 0x83 = reply topic not in own $reply/<clientID>/ namespace.
	if subAck.Reasons[0] == 0x83 {
		log.Fatal("SUBACK 0x83: reply topic must be $reply/<own-clientID>/...")
	}

	// 2. Publish the command with ResponseTopic + CorrelationData.
	pubAck, err := client.Publish(ctx, &paho.Publish{
		Topic:   commandTopic,
		QoS:     1,
		Payload: []byte(`{"action":"restart","target":"service-a"}`),
		Properties: &paho.PublishProperties{
			ResponseTopic:   replyTopic,
			CorrelationData: []byte(correlationID),
		},
	})
	if err != nil {
		log.Fatalf("publish: %v", err)
	}
	// PUBACK is immediate — the execution outcome arrives later on the reply topic.
	if pubAck.ReasonCode != 0 {
		log.Fatalf("PUBACK reason=0x%02X (0x83=bad ResponseTopic, 0x97=RpcMaxPending)", pubAck.ReasonCode)
	}

	// 3. Wait for the response (empty body; kubemq-executed user property).
	select {
	case r := <-resp:
		var executed, errMsg string
		if r.Properties != nil {
			for _, up := range r.Properties.User {
				switch up.Key {
				case "kubemq-executed":
					executed = up.Value
				case "kubemq-error":
					errMsg = up.Value
				}
			}
		}
		if executed == "true" {
			fmt.Println("command executed successfully")
		} else {
			fmt.Printf("command not executed: %s\n", errMsg)
		}
	case <-ctx.Done():
		log.Fatal("timed out — is a gRPC responder running on channel demo.cmd?")
	}
	_ = client.Disconnect(&paho.Disconnect{ReasonCode: 0})
}

import os
import threading
import time
import uuid

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

CLIENT_ID = "py-command-client"
COMMAND_TOPIC = "commands/demo/cmd"            # channel demo.cmd
REPLY_TOPIC = f"$reply/{CLIENT_ID}/inbox"      # own namespace only
CORRELATION = uuid.uuid4().bytes


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"))
    response = threading.Event()
    props: dict[str, str] = {}

    def on_message(client, userdata, msg):
        if msg.properties and hasattr(msg.properties, "UserProperty"):
            for k, v in (msg.properties.UserProperty or []):
                props[k] = v
        response.set()

    client = mqtt.Client(CallbackAPIVersion.VERSION2, client_id=CLIENT_ID, protocol=mqtt.MQTTv5)
    # 1. Subscribe to your own reply topic BEFORE publishing.
    client.on_connect = lambda c, *_: c.subscribe(REPLY_TOPIC, qos=1)
    client.on_message = on_message
    client.connect(host, port, keepalive=30)
    client.loop_start()
    time.sleep(0.5)

    # 2. Publish the command with ResponseTopic + CorrelationData.
    pub_props = Properties(PacketTypes.PUBLISH)
    pub_props.ResponseTopic = REPLY_TOPIC
    pub_props.CorrelationData = CORRELATION
    payload = b'{"action": "restart", "target": "service-a"}'
    # PUBACK is immediate — the execution outcome arrives later on the reply topic.
    client.publish(COMMAND_TOPIC, payload=payload, qos=1, properties=pub_props).wait_for_publish(10)

    # 3. Wait for the response (empty body; kubemq-executed user property).
    if not response.wait(timeout=30):
        raise TimeoutError("no response — is a gRPC responder running on channel demo.cmd?")
    if props.get("kubemq-executed") == "true":
        print("command executed successfully")
    else:
        print(f"command not executed: {props.get('kubemq-error', '(no detail)')}")

    client.loop_stop()
    client.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");
        String clientId = "java-cmd-" + UUID.randomUUID().toString().substring(0, 8);
        String commandTopic = "commands/demo/cmd";          // channel demo.cmd
        String replyTopic = "$reply/" + clientId + "/inbox"; // own namespace only
        byte[] correlation = ("cmd-" + UUID.randomUUID()).getBytes(StandardCharsets.UTF_8);

        CountDownLatch latch = new CountDownLatch(1);
        String[] executed = {null}, error = {null};

        MqttAsyncClient client = new MqttAsyncClient(broker, clientId);
        client.setCallback(new MqttCallback() {
            public void messageArrived(String topic, MqttMessage m) {
                MqttProperties props = m.getProperties();
                if (props != null && props.getUserProperties() != null) {
                    for (UserProperty up : props.getUserProperties()) {
                        if ("kubemq-executed".equals(up.getKey())) executed[0] = up.getValue();
                        if ("kubemq-error".equals(up.getKey())) error[0] = up.getValue();
                    }
                }
                latch.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) {}
        });

        MqttConnectionOptions opts = new MqttConnectionOptions();
        opts.setCleanStart(true);
        opts.setKeepAliveInterval(30);
        client.connect(opts).waitForCompletion(5_000);

        // 1. Subscribe to your own reply topic BEFORE publishing.
        client.subscribe(replyTopic, 1).waitForCompletion(5_000);
        Thread.sleep(200);

        // 2. Publish the command with ResponseTopic + CorrelationData.
        MqttProperties pubProps = new MqttProperties();
        pubProps.setResponseTopic(replyTopic);
        pubProps.setCorrelationData(correlation);
        pubProps.setUserProperties(List.of(new UserProperty("language", "java")));
        MqttMessage msg = new MqttMessage("execute-action".getBytes(StandardCharsets.UTF_8));
        msg.setQos(1);
        msg.setProperties(pubProps);
        // PUBACK is immediate — the execution outcome arrives later on the reply topic.
        client.publish(commandTopic, msg).waitForCompletion(5_000);

        // 3. Wait for the response (empty body; kubemq-executed user property).
        if (!latch.await(15, TimeUnit.SECONDS)) {
            throw new IllegalStateException("no response — is a gRPC responder running on channel demo.cmd?");
        }
        if ("true".equals(executed[0])) {
            System.out.println("command executed successfully");
        } else {
            System.out.printf("command not executed: %s%n", error[0]);
        }

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

import mqtt, { type MqttClient } from "mqtt";
import crypto from "node:crypto";

const CLIENT_ID = `js-cmd-${crypto.randomBytes(4).toString("hex")}`;
const REPLY_TOPIC = `$reply/${CLIENT_ID}/inbox`; // own namespace only
const COMMAND_TOPIC = "commands/demo/cmd";        // channel demo.cmd
const CORRELATION_DATA = Buffer.from(crypto.randomUUID());
const RPC_TIMEOUT_MS = 30_000;

async function main(): Promise<void> {
  const url = process.env["KUBEMQ_MQTT_URL"] ?? "tcp://localhost:1883";
  const client: MqttClient = mqtt.connect(url, {
    clientId: CLIENT_ID,
    protocolVersion: 5,
    clean: true,
    keepalive: 30,
  });
  await new Promise<void>((res, rej) => {
    client.once("connect", () => res());
    client.once("error", rej);
  });

  // 1. Subscribe to your own reply topic BEFORE publishing.
  await new Promise<void>((resolve, reject) => {
    client.subscribe(REPLY_TOPIC, { qos: 1 }, (err, granted) => {
      if (err) return reject(err);
      // 0x83 = reply topic not in own $reply/<clientID>/ namespace.
      if (((granted?.[0]?.qos as number) ?? -1) > 2) return reject(new Error("reply subscribe rejected"));
      resolve();
    });
  });

  // 2. Arm the response handler, then publish with ResponseTopic + CorrelationData.
  const responded = new Promise<void>((resolve, reject) => {
    const timer = setTimeout(
      () => reject(new Error("RPC timeout — is a gRPC responder running on channel demo.cmd?")),
      RPC_TIMEOUT_MS,
    );
    client.on("message", (topic, payload, packet) => {
      if (topic !== REPLY_TOPIC) return;
      clearTimeout(timer);
      const userProps = (packet.properties as { userProperties?: Record<string, string> } | undefined)
        ?.userProperties ?? {};
      // Command responses have an empty body; the result is in the user properties.
      if (userProps["kubemq-executed"] === "true") {
        console.log("command executed successfully");
      } else {
        console.log(`command not executed: ${userProps["kubemq-error"] ?? "(no detail)"}`);
      }
      resolve();
    });
  });

  await new Promise<void>((resolve, reject) => {
    client.publish(
      COMMAND_TOPIC,
      JSON.stringify({ action: "restart", target: "service-a" }),
      { qos: 1, properties: { responseTopic: REPLY_TOPIC, correlationData: CORRELATION_DATA } },
      (err) => (err ? reject(err) : resolve()), // PUBACK is immediate, not the response
    );
  });

  await responded;
  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) 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();
var clientId = $"csharp-cmd-{Guid.NewGuid():N}"[..22];
var requestId = Guid.NewGuid().ToString("N");
var replyTopic = $"$reply/{clientId}/inbox"; // own namespace only
const string commandTopic = "commands/demo/cmd"; // channel demo.cmd

var factory = new MqttFactory();
var responded = new TaskCompletionSource<(bool executed, string? error)>(
    TaskCreationOptions.RunContinuationsAsynchronously);

using var client = factory.CreateMqttClient();
client.ApplicationMessageReceivedAsync += e =>
{
    string? executed = null, error = null;
    foreach (var p in e.ApplicationMessage.UserProperties ?? new())
    {
        if (p.Name == "kubemq-executed") executed = p.Value;
        if (p.Name == "kubemq-error") error = p.Value;
    }
    responded.TrySetResult((executed == "true", error));
    return Task.CompletedTask;
};

var options = new MqttClientOptionsBuilder()
    .WithTcpServer(host, port)
    .WithClientId(clientId)
    .WithProtocolVersion(MQTTnet.Formatter.MqttProtocolVersion.V500)
    .WithCleanSession(true)
    .Build();
await client.ConnectAsync(options);

// 1. Subscribe to your own reply topic BEFORE publishing.
var subResult = await client.SubscribeAsync(new MqttClientSubscribeOptionsBuilder()
    .WithTopicFilter(f => f.WithTopic(replyTopic).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce))
    .Build());
if (subResult.Items.First().ResultCode > MqttClientSubscribeResultCode.GrantedQoS2)
    throw new Exception("reply subscribe rejected — use $reply/<own-clientID>/ namespace");
await Task.Delay(200);

// 2. Publish the command with ResponseTopic + CorrelationData.
var pubResult = await client.PublishAsync(new MqttApplicationMessageBuilder()
    .WithTopic(commandTopic)
    .WithPayload(Encoding.UTF8.GetBytes($"{{\"action\":\"restart\",\"requestId\":\"{requestId}\"}}"))
    .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
    .WithResponseTopic(replyTopic)
    .WithCorrelationData(Encoding.UTF8.GetBytes(requestId))
    .Build());
// PUBACK is immediate — the execution outcome arrives later on the reply topic.
if (pubResult.ReasonCode != MqttClientPublishReasonCode.Success)
    throw new Exception($"PUBACK reason: {pubResult.ReasonCode}");

// 3. Wait for the response (empty body; kubemq-executed user property).
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
var (executed, error) = await responded.Task.WaitAsync(cts.Token);
Console.WriteLine(executed
    ? "command executed successfully"
    : $"command not executed: {error ?? "(no detail)"}");

await client.DisconnectAsync();

use bytes::Bytes;
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 client_id = format!("rust-cmd-{}", &Uuid::new_v4().to_string()[..8]);
    let reply_topic = format!("$reply/{}/inbox", client_id); // own namespace only
    let command_topic = "commands/demo/cmd";                  // channel demo.cmd
    let correlation = Uuid::new_v4().to_string();
    let corr_bytes = Bytes::from(correlation.clone().into_bytes());

    let mut opts = MqttOptions::new(&client_id, &host, port);
    opts.set_keep_alive(Duration::from_secs(30));
    let (client, mut eventloop) = AsyncClient::new(opts, 10);

    let (tx, rx) = oneshot::channel::<(bool, Option<String>)>();
    let reply_clone = reply_topic.clone();
    let corr_check = corr_bytes.clone();
    tokio::spawn(async move {
        let mut tx = Some(tx);
        loop {
            match eventloop.poll().await {
                Ok(Event::Incoming(Packet::Publish(p))) => {
                    if String::from_utf8_lossy(&p.topic) != reply_clone {
                        continue;
                    }
                    let props = p.properties.as_ref();
                    // Match the response to the request via CorrelationData.
                    if props.and_then(|pr| pr.correlation_data.clone()).as_deref()
                        != Some(corr_check.as_ref())
                    {
                        continue;
                    }
                    // Command responses have an empty body; the result is in the user properties.
                    let executed = props
                        .and_then(|pr| pr.user_properties.iter().find(|(k, _)| k == "kubemq-executed"))
                        .map(|(_, v)| v == "true")
                        .unwrap_or(false);
                    let error = props
                        .and_then(|pr| pr.user_properties.iter().find(|(k, _)| k == "kubemq-error"))
                        .map(|(_, v)| v.clone());
                    if let Some(tx) = tx.take() {
                        let _ = tx.send((executed, error));
                        return;
                    }
                }
                Ok(_) => {}
                Err(e) => { eprintln!("event loop: {e}"); return; }
            }
        }
    });

    // 1. Subscribe to your own reply topic BEFORE publishing.
    client.subscribe(&reply_topic, QoS::AtLeastOnce).await?;
    tokio::time::sleep(Duration::from_millis(300)).await;

    // 2. Publish the command with ResponseTopic + CorrelationData.
    let props = PublishProperties {
        response_topic: Some(reply_topic.clone()),
        correlation_data: Some(corr_bytes),
        ..Default::default()
    };
    // PUBACK is immediate — the execution outcome arrives later on the reply topic.
    client
        .publish_with_properties(
            command_topic,
            QoS::AtLeastOnce,
            false,
            br#"{"action":"restart","target":"service-a"}"#.as_ref(),
            props,
        )
        .await?;

    // 3. Wait for the response (empty body; kubemq-executed user property).
    let (executed, error) = timeout(Duration::from_secs(30), rx)
        .await
        .map_err(|_| "timed out — is a gRPC responder running on channel demo.cmd?")??;
    if executed {
        println!("command executed successfully");
    } else {
        println!("command not executed: {}", error.as_deref().unwrap_or("(no detail)"));
    }
    Ok(())
}