# Queries (/connectors/stomp/how-to/queries) Queries are **request/reply RPC that returns a payload** over the STOMP connector. A client SENDs to a `/query/` destination and receives a single reply **MESSAGE** — carrying the responder's **response body and tags** — on a connection-local `/reply/...` subscription. A Query is the "execute this and give me the answer" half of KubeMQ RPC; the "execute this and confirm it worked" half (typically no body) is [Commands](/connectors/stomp/how-to/commands). Queries use the **exact same 3-step flow and `stomp-error` failure mode** as Commands — the only differences are the destination prefix (`/query/` vs `/command/`) and that a Query's reply carries a **response body + tags**. See [Commands](/connectors/stomp/how-to/commands) for the canonical write-up of the flow, the `reply-to` / `correlation-id` / `timeout` headers, and the three `stomp-error` failure shapes; this page focuses on the Query-specific reply payload. **STOMP is RPC-requester-only.** A STOMP client can **send** queries but cannot **respond** to them — the responder runs on the gRPC side, using a native KubeMQ SDK. A `SUBSCRIBE /query/...` is **hard-rejected** with `ERROR "cannot subscribe to RPC destinations"` and the connection closes. To answer queries, run a responder with the gRPC SDK (`SubscribeToQueries` + `SendQueryResponse`); the connector bridges your STOMP SEND to it. ## Overview [#overview] The `/query/` prefix selects the Queries pattern; `/queries/` is an accepted alias that egress canonicalizes back to `/query/...` — **lead with `/query/` in your code**. The remaining segments are slash-to-dot joined into the KubeMQ channel: `/query/users/lookup` → channel `users.lookup`. `/reply/` is the connection-local reply destination — no alias, authz-exempt, 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 query | `SEND /query/` with `reply-to` | `SendQuery` (dispatched to the gRPC-side responder) | | Receive the answer | `MESSAGE` on `/reply/` | the responder's **response body + tags** | ## How it works [#how-it-works] The flow is identical to [Commands](/connectors/stomp/how-to/commands): subscribe to a reply inbox first, SEND the query with a required `reply-to` header, and the answer arrives back as a MESSAGE on that inbox — except a Query's reply carries the responder's **response body**, and its response **tags** surface as MESSAGE **headers** (including `content-type` via the `stomp.*` mapping). *A Query reply carries a response payload; the responder runs on the gRPC side and the connector bridges the two.* ## The 3-step flow with a response payload [#the-3-step-flow-with-a-response-payload] 1. **SUBSCRIBE to `/reply/` first** (connection-local: no array, no authz, no ack tracking). It must be active on the **same connection** before you SEND. 2. **SEND to `/query/`** with the **required** `reply-to` (the `/reply/` from step 1, same connection — missing or inactive → `ERROR "reply-to subscription required"` and close), an optional `correlation-id` (echoed back only when set), and an optional `timeout` in **milliseconds** (effective `min(timeout, server cap)`, default 30000). 3. **The reply MESSAGE on `/reply/`** carries: * the responder's **response body** (the payload — this is what distinguishes a Query from a Command), * the responder's **response tags**, surfaced as MESSAGE **headers** (including `content-type`), * `correlation-id` echoed only when the request set it, * `destination` = the reply-to, a fresh `message-id`, and `subscription` = the reply sub id (1.1/1.2 only). ## Failures are a MESSAGE with a `stomp-error` header [#failures-are-a-message-with-a-stomp-error-header] The failure mode is **identical to Commands**. **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. For a Query this is especially relevant: a Query reply normally has a body, so a **logical error** carries `stomp-error` **and** the responder's body + tags; only a **transport error / timeout** (`stomp-error` + empty body, `context deadline exceeded` sanitized to `timeout`) and a **nil response** (`stomp-error:"no response"` + empty body) are empty-bodied. Only `reply-to` violations and pending-cap overflow close the connection. ## Send a query [#send-a-query] Each example performs the 3-step requester flow against a Queries responder running on the gRPC side. It subscribes to a reply inbox, SENDs the query with `reply-to` + `correlation-id` + `timeout`, then reads the response body (checking the `stomp-error` header first 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/q1" // connection-local reply inbox queryDest = "/query/lookup" // Queries pattern → channel lookup ) 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/q1 FIRST (connection-local). reply, err := conn.Subscribe(replyDest, stomp.AckAuto) if err != nil { log.Fatalf("subscribe reply: %v", err) } // Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms). if err := conn.Send(queryDest, "application/json", []byte(`{"key":"user-42"}`), stomp.SendOpt.Header("reply-to", replyDest), stomp.SendOpt.Header("correlation-id", "abc123"), stomp.SendOpt.Header("timeout", "5000"), ); err != nil { log.Fatalf("send query: %v", err) } // Step 3: receive the response body on /reply/q1. select { case msg := <-reply.C: if se := msg.Header.Get("stomp-error"); se != "" { log.Fatalf("query failed: stomp-error=%q (connection stays open)", se) } fmt.Printf("query answered: %s (content-type=%s)\n", string(msg.Body), msg.Header.Get("content-type")) 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/q1" # connection-local reply inbox QUERY_DEST = "/query/lookup" # Queries pattern → channel lookup 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/q1 FIRST (connection-local). conn.subscribe(REPLY_DEST, id="q1", ack="auto") # Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms). conn.send(QUERY_DEST, '{"key":"user-42"}', content_type="application/json", headers={"reply-to": REPLY_DEST, "correlation-id": "abc123", "timeout": "5000"}) # Step 3: receive the response body on /reply/q1. frame = replies.inbox.get(timeout=10) if frame.headers.get("stomp-error"): raise SystemExit(f"query failed: {frame.headers['stomp-error']!r}") print(f"query answered: {frame.body} (content-type={frame.headers.get('content-type')})") 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/q1"; // connection-local reply inbox private static final String QUERY_DEST = "/query/lookup"; // Queries pattern → channel lookup 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/q1 FIRST (connection-local). StompHeaders replyHeaders = new StompHeaders(); replyHeaders.setDestination(REPLY_DEST); replyHeaders.setId("q1"); replyHeaders.setAck("auto"); conn.subscribe(replyHeaders, new StompSessionHandlerAdapter() { @Override public Type getPayloadType(StompHeaders headers) { return String.class; } @Override public void handleFrame(StompHeaders headers, Object payload) { inbox.add(new Object[] { headers, payload }); } }); // Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms). StompHeaders q = new StompHeaders(); q.setDestination(QUERY_DEST); q.add("content-type", "application/json"); q.add("reply-to", REPLY_DEST); q.add("correlation-id", "abc123"); q.add("timeout", "5000"); conn.send(q, "{\"key\":\"user-42\"}".getBytes()); // Step 3: receive the response body on /reply/q1. Object[] reply = inbox.poll(10, TimeUnit.SECONDS); if (reply == null) throw new IllegalStateException("timed out waiting for the reply"); StompHeaders headers = (StompHeaders) reply[0]; if (headers.getFirst("stomp-error") != null) { throw new IllegalStateException("query failed: " + headers.getFirst("stomp-error")); } System.out.printf("query answered: %s (content-type=%s)%n", reply[1], headers.getFirst("content-type")); conn.disconnect(); client.stop(); } } ``` ```typescript import { connect, type Client } from "stompit"; const REPLY_DEST = "/reply/q1"; // connection-local reply inbox const QUERY_DEST = "/query/lookup"; // Queries pattern → channel lookup 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/q1 FIRST (connection-local). const reply = new Promise<{ headers: Record; body: string }>((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, body) => { clearTimeout(timer); if (readErr) return reject(readErr); resolve({ headers: message.headers as Record, body: body ?? "" }); }); }); }); // Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms). const frame = conn.send({ destination: QUERY_DEST, "content-type": "application/json", "reply-to": REPLY_DEST, "correlation-id": "abc123", timeout: "5000", }); frame.write(JSON.stringify({ key: "user-42" })); frame.end(); // Step 3: inspect the response body on /reply/q1. const { headers, body } = await reply; if (headers["stomp-error"]) throw new Error(`query failed: ${headers["stomp-error"]}`); console.log(`query answered: ${body} (content-type=${headers["content-type"]})`); 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/q1"; // connection-local reply inbox const string queryDest = "/query/lookup"; // Queries pattern → channel lookup 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/q1 FIRST (connection-local). using var replyConsumer = session.CreateConsumer(session.GetQueue(replyDest)); // Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms). using var producer = session.CreateProducer(session.GetQueue(queryDest)); var q = producer.CreateBytesMessage(Encoding.UTF8.GetBytes("{\"key\":\"user-42\"}")); q.StompType = "application/json"; q.Headers.SetValue("reply-to", replyDest); q.Headers.SetValue("correlation-id", "abc123"); q.Headers.SetValue("timeout", "5000"); producer.Send(q); // Step 3: receive the response body on /reply/q1. 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($"query failed: {stompError}"); Console.WriteLine($"query answered: {Encoding.UTF8.GetString(reply.Content)} " + $"(content-type={reply.Headers.GetValue("content-type")})"); ``` ```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/q1" # connection-local reply inbox QUERY_DEST = "/query/lookup" # Queries pattern → channel lookup conn = Stomp::Client.new(hosts: hosts, connect_headers: { "accept-version" => "1.2", "host" => "/" }) # Step 1: SUBSCRIBE /reply/q1 FIRST (connection-local). inbox = Thread::Queue.new conn.subscribe(REPLY_DEST, id: "q1", ack: "auto") { |msg| inbox << msg } # Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms). conn.publish(QUERY_DEST, '{"key":"user-42"}', "content-type" => "application/json", "reply-to" => REPLY_DEST, "correlation-id" => "abc123", "timeout" => "5000") # Step 3: receive the response body on /reply/q1. reply = Timeout.timeout(10) { inbox.pop } raise "query failed: #{reply.headers['stomp-error']}" if reply.headers["stomp-error"] puts "query answered: #{reply.body} (content-type=#{reply.headers['content-type']})" 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/q1"; // connection-local reply inbox const QUERY_DEST: &str = "/query/lookup"; // Queries pattern → channel lookup 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/q1 FIRST (connection-local). conn.send(ToServer::Subscribe { destination: REPLY_DEST.into(), id: "q1".into(), ack: Some(AckMode::Auto), }.into()) .await?; // Step 2: SEND /query/lookup with reply-to + correlation-id + timeout(ms). conn.send(ToServer::Send { destination: QUERY_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#"{"key":"user-42"}"#.to_vec()), }.into()) .await?; // Step 3: receive the response body on /reply/q1. let frame = tokio::time::timeout(Duration::from_secs(10), conn.next()) .await? .ok_or("stream closed")??; if let FromServer::Message { headers, body, .. } = frame.content { if let Some((_, err)) = headers.iter().find(|(k, _)| k == "stomp-error") { return Err(format!("query failed: {err}").into()); } let payload = String::from_utf8_lossy(&body.unwrap_or_default()).into_owned(); println!("query answered: {payload}"); } Ok(()) } ``` ## Where the responder lives [#where-the-responder-lives] Because a STOMP client cannot be a responder, the Queries **responder must run on the gRPC side** — a process using a native KubeMQ SDK that does `SubscribeToQueries`, computes the response, and replies via `SendQueryResponse` (the response **body + tags** are what the STOMP requester receives). The connector bridges the STOMP requester's SEND to that responder over the broker, the same path the gRPC connector uses. In-flight RPCs are bounded by a pending cap (default 1024); overflow → `ERROR "too many pending requests"` and the connection closes. ## Related [#related]