# Commands (/connectors/stomp/how-to/commands) Commands are **request/reply RPC** over the STOMP connector. A client SENDs to a `/command/` destination and receives a single reply **MESSAGE** on a connection-local `/reply/...` subscription. A Command is the "execute this and tell me it worked" half of KubeMQ RPC — the responder returns an **execution result** (success or failure), typically without a body. For "execute this and give me the answer" with a response payload, use [Queries](/connectors/stomp/how-to/queries). **STOMP is RPC-requester-only.** A STOMP client can **send** commands but cannot **respond** to them — the responder runs on the gRPC side, using a native KubeMQ SDK. A `SUBSCRIBE /command/...` is **hard-rejected** with `ERROR "cannot subscribe to RPC destinations"` and the connection closes. To answer commands, run a responder with the gRPC SDK (`SubscribeToCommands` + `SendCommandResponse`); the connector bridges your STOMP SEND to it. ## Overview [#overview] The `/command/` prefix selects the Commands pattern; `/commands/` is an accepted alias that egress canonicalizes back to `/command/...` — **lead with `/command/` in your code**. The remaining segments are slash-to-dot joined into the KubeMQ channel: `/command/orders/exec` → channel `orders.exec`. `/reply/` is the connection-local reply destination — it has **no alias**, is authz-exempt, and is never an array subscription. | Operation | STOMP action | KubeMQ mapping | | -------------------- | ------------------------------------- | ----------------------------------------------------- | | Subscribe to replies | `SUBSCRIBE /reply/` | connection-local inbox (no array, no authz, no ack) | | Send a command | `SEND /command/` with `reply-to` | `SendCommand` (dispatched to the gRPC-side responder) | | Receive the result | `MESSAGE` on `/reply/` | the responder's execution result | ## How it works [#how-it-works] The requester subscribes to a reply inbox first, SENDs the command with a required `reply-to` header, and the execution result arrives back as a MESSAGE on that inbox. The responder is a separate process on the gRPC side. *The STOMP client is the requester only; the responder runs on the gRPC side and the connector bridges the two.* ## The 3-step flow [#the-3-step-flow] 1. **Step 1: SUBSCRIBE to `/reply/` first.** The reply destination is connection-local: no array subscription, no authz (it is authz-exempt), no ack tracking. It must be active on the **same connection** before you SEND, or the SEND is rejected. 2. **Step 2: SEND to `/command/`** with these headers: | Header | Required | Meaning | | ---------------- | -------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `reply-to` | **yes** | the `/reply/` you subscribed to in step 1, on the **same** connection. Missing or inactive → `ERROR "reply-to subscription required"` and the connection closes. | | `correlation-id` | no | echoed back **verbatim** on the reply — **but only when the request set it**. | | `timeout` | no | in **milliseconds**; effective = `min(timeout, server cap)`; garbage / non-positive / over-cap → the default (30000). | 3. **Step 3: the reply arrives as a MESSAGE on `/reply/`** with `destination` = the reply-to, a fresh `message-id`, `subscription` = the reply sub id (1.1/1.2 only), and `correlation-id` echoed only when the request set it. ## Failures are a MESSAGE with a `stomp-error` header [#failures-are-a-message-with-a-stomp-error-header] This is the single most surprising RPC behavior — drill it into your client code. **RPC failures are a MESSAGE + `stomp-error` header, NOT an ERROR frame.** A timeout, a logical error, or a dropped reply arrives as **data on the `/reply/` subscription**, and the **connection stays open**. Detect a failure by the **presence of the `stomp-error` header** — not by an ERROR frame, and not by an empty body. The body shape differs by failure kind: a **logical error** (the responder ran but reported failure) carries `stomp-error` **and** the responder's body + tags; a **transport error / timeout** carries `stomp-error` and an **empty body** (`context deadline exceeded` is sanitized to `timeout`); a **nil response** carries `stomp-error:"no response"` and an empty body. Only `reply-to` violations and pending-cap overflow close the connection. ## Send a command [#send-a-command] Each example performs the 3-step requester flow against a Commands responder running on the gRPC side. It subscribes to a reply inbox, SENDs the command with `reply-to` + `correlation-id` + `timeout`, and reads the execution result — checking the `stomp-error` header to distinguish success from failure. Every client reads the connector endpoint from `KUBEMQ_STOMP_URL` (default `tcp://localhost:61613`). ```go package main import ( "fmt" "log" "net/url" "os" "time" "github.com/go-stomp/stomp/v3" ) const ( replyDest = "/reply/r1" // connection-local reply inbox commandDest = "/command/exec" // Commands pattern → channel exec ) func addr() (network, host string) { u, _ := url.Parse(os.Getenv("KUBEMQ_STOMP_URL")) if u == nil || u.Host == "" { return "tcp", "localhost:61613" } return "tcp", u.Host } func main() { network, host := addr() conn, err := stomp.Dial(network, host) if err != nil { log.Fatalf("dial: %v", err) } defer func() { _ = conn.Disconnect() }() // Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local). reply, err := conn.Subscribe(replyDest, stomp.AckAuto) if err != nil { log.Fatalf("subscribe reply: %v", err) } // Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms). corrID := "abc123" if err := conn.Send(commandDest, "application/json", []byte(`{"op":"do-work"}`), stomp.SendOpt.Header("reply-to", replyDest), stomp.SendOpt.Header("correlation-id", corrID), stomp.SendOpt.Header("timeout", "5000"), ); err != nil { log.Fatalf("send command: %v", err) } // Step 3: receive the execution result on /reply/r1. select { case msg := <-reply.C: if se := msg.Header.Get("stomp-error"); se != "" { log.Fatalf("command failed: stomp-error=%q (connection stays open)", se) } fmt.Printf("command executed (correlation-id=%s)\n", msg.Header.Get("correlation-id")) case <-time.After(10 * time.Second): log.Fatal("timed out waiting for the reply") } } ``` ```python import os import queue from urllib.parse import urlparse import stomp REPLY_DEST = "/reply/r1" # connection-local reply inbox COMMAND_DEST = "/command/exec" # Commands pattern → channel exec def endpoint() -> tuple[str, int]: parsed = urlparse(os.environ.get("KUBEMQ_STOMP_URL", "tcp://localhost:61613")) return parsed.hostname or "localhost", parsed.port or 61613 class Replies(stomp.ConnectionListener): def __init__(self) -> None: self.inbox: queue.Queue = queue.Queue() def on_message(self, frame) -> None: self.inbox.put(frame) def main() -> None: host, port = endpoint() replies = Replies() conn = stomp.Connection12([(host, port)], heartbeats=(10000, 10000)) conn.set_listener("r", replies) conn.connect(wait=True) # Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local). conn.subscribe(REPLY_DEST, id="r1", ack="auto") # Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms). conn.send(COMMAND_DEST, '{"op":"do-work"}', content_type="application/json", headers={"reply-to": REPLY_DEST, "correlation-id": "abc123", "timeout": "5000"}) # Step 3: receive the execution result on /reply/r1. frame = replies.inbox.get(timeout=10) if frame.headers.get("stomp-error"): raise SystemExit(f"command failed: {frame.headers['stomp-error']!r}") print(f"command executed (correlation-id={frame.headers.get('correlation-id')})") conn.disconnect() if __name__ == "__main__": main() ``` ```java import java.lang.reflect.Type; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeUnit; import org.springframework.messaging.simp.stomp.StompHeaders; import org.springframework.messaging.simp.stomp.StompSession; import org.springframework.messaging.simp.stomp.StompSessionHandlerAdapter; import org.springframework.messaging.simp.stomp.ReactorNettyTcpStompClient; public final class Main { private static final String REPLY_DEST = "/reply/r1"; // connection-local reply inbox private static final String COMMAND_DEST = "/command/exec"; // Commands pattern → channel exec public static void main(String[] args) throws Exception { String url = System.getenv().getOrDefault("KUBEMQ_STOMP_URL", "tcp://localhost:61613"); java.net.URI u = java.net.URI.create(url); ReactorNettyTcpStompClient client = new ReactorNettyTcpStompClient(u.getHost(), u.getPort() > 0 ? u.getPort() : 61613); BlockingQueue inbox = new ArrayBlockingQueue<>(1); StompSession conn = client.connectAsync(new StompSessionHandlerAdapter() {}).get(10, TimeUnit.SECONDS); // Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local). StompHeaders replyHeaders = new StompHeaders(); replyHeaders.setDestination(REPLY_DEST); replyHeaders.setId("r1"); replyHeaders.setAck("auto"); conn.subscribe(replyHeaders, new StompSessionHandlerAdapter() { @Override public Type getPayloadType(StompHeaders headers) { return byte[].class; } @Override public void handleFrame(StompHeaders headers, Object payload) { inbox.add(headers); } }); // Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms). StompHeaders cmd = new StompHeaders(); cmd.setDestination(COMMAND_DEST); cmd.add("content-type", "application/json"); cmd.add("reply-to", REPLY_DEST); cmd.add("correlation-id", "abc123"); cmd.add("timeout", "5000"); conn.send(cmd, "{\"op\":\"do-work\"}".getBytes()); // Step 3: receive the execution result on /reply/r1. StompHeaders reply = inbox.poll(10, TimeUnit.SECONDS); if (reply == null) throw new IllegalStateException("timed out waiting for the reply"); if (reply.getFirst("stomp-error") != null) { throw new IllegalStateException("command failed: " + reply.getFirst("stomp-error")); } System.out.printf("command executed (correlation-id=%s)%n", reply.getFirst("correlation-id")); conn.disconnect(); client.stop(); } } ``` ```typescript import { connect, type Client } from "stompit"; const REPLY_DEST = "/reply/r1"; // connection-local reply inbox const COMMAND_DEST = "/command/exec"; // Commands pattern → channel exec function endpoint(): { host: string; port: number } { const url = new URL(process.env["KUBEMQ_STOMP_URL"] ?? "tcp://localhost:61613"); return { host: url.hostname, port: Number(url.port) || 61613 }; } function open(): Promise { const { host, port } = endpoint(); return new Promise((resolve, reject) => { connect({ host, port, connectHeaders: { "accept-version": "1.2", "heart-beat": "10000,10000" } }, (err, client) => (err ? reject(err) : resolve(client))); }); } async function main(): Promise { const conn = await open(); // Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local). const reply = new Promise>((resolve, reject) => { const timer = setTimeout(() => reject(new Error("timed out waiting for the reply")), 15_000); conn.subscribe({ destination: REPLY_DEST, ack: "auto" }, (err, message) => { if (err) return reject(err); message.readString("utf-8", (readErr) => { clearTimeout(timer); if (readErr) return reject(readErr); resolve(message.headers as Record); }); }); }); // Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms). const frame = conn.send({ destination: COMMAND_DEST, "content-type": "application/json", "reply-to": REPLY_DEST, "correlation-id": "abc123", timeout: "5000", }); frame.write(JSON.stringify({ op: "do-work" })); frame.end(); // Step 3: inspect the execution result on /reply/r1. const headers = await reply; if (headers["stomp-error"]) throw new Error(`command failed: ${headers["stomp-error"]}`); console.log(`command executed (correlation-id=${headers["correlation-id"]})`); await new Promise((r) => conn.disconnect(() => r())); } main().catch((err) => { console.error(err); process.exit(1); }); ``` ```csharp using System.Text; using Stomp.Net; var url = Environment.GetEnvironmentVariable("KUBEMQ_STOMP_URL") ?? "tcp://localhost:61613"; var uri = new Uri(url); const string replyDest = "/reply/r1"; // connection-local reply inbox const string commandDest = "/command/exec"; // Commands pattern → channel exec string brokerUri = $"stomp:tcp://{uri.Host}:{(uri.Port > 0 ? uri.Port : 61613)}"; var factory = new ConnectionFactory(brokerUri) { UserName = "kubemq", Password = "" }; using var conn = factory.CreateConnection(); conn.Start(); using var session = conn.CreateSession(AcknowledgementMode.AutoAcknowledge); // Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local). using var replyConsumer = session.CreateConsumer(session.GetQueue(replyDest)); // Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms). using var producer = session.CreateProducer(session.GetQueue(commandDest)); var cmd = producer.CreateBytesMessage(Encoding.UTF8.GetBytes("{\"op\":\"do-work\"}")); cmd.StompType = "application/json"; cmd.Headers.SetValue("reply-to", replyDest); cmd.Headers.SetValue("correlation-id", "abc123"); cmd.Headers.SetValue("timeout", "5000"); producer.Send(cmd); // Step 3: receive the execution result on /reply/r1. var reply = replyConsumer.Receive(TimeSpan.FromSeconds(10)) ?? throw new InvalidOperationException("timed out waiting for the reply"); var stompError = reply.Headers.GetValue("stomp-error"); if (!string.IsNullOrEmpty(stompError)) throw new InvalidOperationException($"command failed: {stompError}"); Console.WriteLine($"command executed (correlation-id={reply.Headers.GetValue("correlation-id")})"); ``` ```ruby require "stomp" require "uri" require "timeout" uri = URI.parse(ENV.fetch("KUBEMQ_STOMP_URL", "tcp://localhost:61613")) hosts = [{ host: uri.host, port: uri.port || 61613, login: "kubemq", passcode: "" }] REPLY_DEST = "/reply/r1" # connection-local reply inbox COMMAND_DEST = "/command/exec" # Commands pattern → channel exec conn = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" }) # Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local). inbox = Thread::Queue.new conn.subscribe(REPLY_DEST, id: "r1", ack: "auto") { |msg| inbox << msg } # Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms). conn.publish(COMMAND_DEST, '{"op":"do-work"}', "content-type" => "application/json", "reply-to" => REPLY_DEST, "correlation-id" => "abc123", "timeout" => "5000") # Step 3: receive the execution result on /reply/r1. reply = Timeout.timeout(10) { inbox.pop } raise "command failed: #{reply.headers['stomp-error']}" if reply.headers["stomp-error"] puts "command executed (correlation-id=#{reply.headers['correlation-id']})" conn.close ``` ```rust use std::time::Duration; use async_stomp::client::Connector; use async_stomp::{AckMode, FromServer, ToServer}; use futures::{SinkExt, StreamExt}; const REPLY_DEST: &str = "/reply/r1"; // connection-local reply inbox const COMMAND_DEST: &str = "/command/exec"; // Commands pattern → channel exec fn host_port() -> (String, u16) { let url = std::env::var("KUBEMQ_STOMP_URL").unwrap_or_else(|_| "tcp://localhost:61613".into()); let hp = url.trim_start_matches("tcp://").trim_start_matches("tls://"); let mut parts = hp.splitn(2, ':'); let host = parts.next().unwrap_or("localhost").to_string(); let port = parts.next().and_then(|p| p.parse().ok()).unwrap_or(61613); (host, port) } #[tokio::main] async fn main() -> Result<(), Box> { let (host, port) = host_port(); let mut conn = Connector::builder() .server(format!("{host}:{port}")) .virtualhost(&host) .connect() .await?; // Step 1: SUBSCRIBE /reply/r1 FIRST (connection-local). conn.send(ToServer::Subscribe { destination: REPLY_DEST.into(), id: "r1".into(), ack: Some(AckMode::Auto), }.into()) .await?; // Step 2: SEND /command/exec with reply-to + correlation-id + timeout(ms). conn.send(ToServer::Send { destination: COMMAND_DEST.into(), transaction: None, headers: Some(vec![ ("content-type".into(), "application/json".into()), ("reply-to".into(), REPLY_DEST.into()), ("correlation-id".into(), "abc123".into()), ("timeout".into(), "5000".into()), ]), body: Some(br#"{"op":"do-work"}"#.to_vec()), }.into()) .await?; // Step 3: receive the execution result on /reply/r1. let frame = tokio::time::timeout(Duration::from_secs(10), conn.next()) .await? .ok_or("stream closed")??; if let FromServer::Message { headers, .. } = frame.content { if let Some((_, err)) = headers.iter().find(|(k, _)| k == "stomp-error") { return Err(format!("command failed: {err}").into()); } let corr = headers.iter().find(|(k, _)| k == "correlation-id").map(|(_, v)| v.as_str()); println!("command executed (correlation-id={})", corr.unwrap_or("")); } Ok(()) } ``` ## Where the responder lives [#where-the-responder-lives] Because a STOMP client cannot be a responder, the Commands **responder must run on the gRPC side** — a process using a native KubeMQ SDK that does `SubscribeToCommands`, processes the request, and replies via `SendCommandResponse`. The connector bridges the STOMP requester's SEND to that responder over the broker, the same path the gRPC connector uses. The number of in-flight RPCs is bounded by a pending cap (default 1024); overflow → `ERROR "too many pending requests"` and the connection closes. ## Related [#related]