# Commands (/connectors/mqtt/how-to/commands) Commands are **RPC with an execution acknowledgement** over the MQTT connector. An MQTT client publishes to `commands/` and a **gRPC-side responder** processes the request and returns a pass/fail result. The MQTT client receives that result on its private reply topic — there is **no body**, only an executed/error status. ## Overview [#overview] The `commands/` prefix selects the Commands pattern (`/` → `.`: `commands/device/reboot` → channel `device.reboot`). The MQTT client is always the **caller**: it subscribes to its own `$reply//` reply topic, then publishes the command with an MQTT 5.0 **response-topic** and **correlation-data**. The responder runs on the **gRPC side** — MQTT clients cannot register as responders. | Step | MQTT action | Notes | | ------------------ | ------------------------------------------------------------- | ----------------------------------------------------------------------- | | Subscribe to reply | `SUBSCRIBE $reply//` | mochi-local; never routed to the broker | | Send command | `PUBLISH commands/` + `ResponseTopic` + `CorrelationData` | `SendCommandRequest` to the gRPC responder | | Receive PUBACK | immediate `PUBACK 0x00` | acks receipt **only**, not execution | | Receive response | `PUBLISH` on the reply topic | empty body; `kubemq-executed` + optional `kubemq-error` user properties | **Commands require MQTT 5.0.** The flow relies on the MQTT 5.0 `ResponseTopic` and `CorrelationData` properties, which MQTT 3.1.1 does not have. A v3.1.1 publish to `commands/` is **silently dropped** — the connector still returns `PUBACK 0x00`, but the message never reaches a responder and no reply arrives. The Ruby `mqtt` gem is 3.1.1-only, so RPC is unavailable from Ruby; the examples below omit it. **`PUBACK` is immediate — it is not the response.** The broker returns `PUBACK` as soon as it receives your publish, before the RPC round-trip completes. You **must** implement your own response-wait timeout on the `$reply` topic; if no responder is registered, no reply ever arrives (the server audits `rpc.timeout` after `RpcTimeoutSeconds`, default 30 s). ## How it works [#how-it-works] The MQTT client subscribes to its reply topic, publishes the command with a response-topic and correlation-data, and waits. The connector bridges the request to a gRPC responder and routes the responder's pass/fail result back to the reply topic. *The MQTT client is the caller; a gRPC responder executes the command and the result returns on the client's `$reply` topic.* ## Request and response [#request-and-response] Each example subscribes to its own `$reply//inbox` topic, publishes a command to `commands/demo/cmd` with a response-topic and correlation-data, then waits for the result and reads the `kubemq-executed` user property. A gRPC responder must be running on channel `demo.cmd`. Every client reads the broker endpoint from `KUBEMQ_MQTT_URL` (default `tcp://localhost:1883`). Ruby is omitted — RPC requires MQTT 5.0 and the `mqtt` gem is 3.1.1-only. ```go 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// namespace. if subAck.Reasons[0] == 0x83 { log.Fatal("SUBACK 0x83: reply topic must be $reply//...") } // 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}) } ``` ```python 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() ``` ```java 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(); } } ``` ```typescript 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 { 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((res, rej) => { client.once("connect", () => res()); client.once("error", rej); }); // 1. Subscribe to your own reply topic BEFORE publishing. await new Promise((resolve, reject) => { client.subscribe(REPLY_TOPIC, { qos: 1 }, (err, granted) => { if (err) return reject(err); // 0x83 = reply topic not in own $reply// 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((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 } | 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((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); }); ``` ```csharp 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// 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(); ``` ```rust 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> { 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)>(); 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(()) } ``` ## The response shape [#the-response-shape] A command response has **no payload** — the outcome is conveyed entirely through MQTT 5.0 user properties: | User property | Value | Meaning | | ----------------- | -------------------- | -------------------------------------------------------- | | `kubemq-executed` | `"true"` / `"false"` | Whether the responder processed the command successfully | | `kubemq-error` | string (optional) | Error message, present only when `kubemq-executed=false` | The connector echoes your `CorrelationData` verbatim on the response — set a unique value (a UUID or sequence number) per request and match the reply on it when you have more than one command in flight. ## Reason codes and responders [#reason-codes-and-responders] | Code | Meaning | | ----------------------- | ----------------------------------------------------------------------------------- | | `PUBACK 0x83` | Missing or foreign `ResponseTopic` (not in your own `$reply//` namespace) | | `PUBACK 0x97` | `RpcMaxPending` quota reached (default 1024 concurrent pending RPCs) | | `PUBACK 0x00` (dropped) | v3.1.1 publish — silently discarded; the message never reaches a responder | | `SUBACK 0x83` | Subscribe to `commands/` — MQTT clients cannot be responders | MQTT clients can only be callers. Responders must use the KubeMQ gRPC API (for example, `SubscribeToCommands` + `SendCommandResponse`). ## Related [#related]