# Queries (/connectors/mqtt/how-to/queries) Queries are **RPC with a data response** over the MQTT connector. An MQTT client publishes to `queries/` and a **gRPC-side responder** returns a payload (body + metadata). The flow is identical to [Commands](/connectors/mqtt/how-to/commands) — subscribe to a reply topic, publish with a response-topic and correlation-data, wait for the reply — the only difference is the reply shape: **a command returns an executed/error status with no body; a query returns an actual data payload.** ## Overview [#overview] The `queries/` prefix selects the Queries pattern (`/` → `.`: `queries/inventory/check` → channel `inventory.check`). The MQTT client is always the **caller**; the responder runs on the **gRPC side** and cannot be an MQTT client. | Step | MQTT action | Notes | | ------------------ | ------------------------------------------------------------ | ------------------------------------------------------- | | Subscribe to reply | `SUBSCRIBE $reply//` | mochi-local; never routed to the broker | | Send query | `PUBLISH queries/` + `ResponseTopic` + `CorrelationData` | `SendQueryRequest` to the gRPC responder | | Receive PUBACK | immediate `PUBACK 0x00` | acks receipt **only**, not the response | | Receive response | `PUBLISH` on the reply topic | **non-empty body** + `kubemq-metadata` + responder tags | | Aspect | Commands | Queries | | ------------------- | --------------------------------- | ---------------------------------- | | MQTT prefix | `commands/` | `queries/` | | Response body | always empty | present — the responder's data | | Response user-props | `kubemq-executed`, `kubemq-error` | `kubemq-metadata` + responder tags | | Use case | "execute this; did it work?" | "give me data" | **Queries require MQTT 5.0.** Like Commands, the flow relies on the MQTT 5.0 `ResponseTopic` and `CorrelationData` properties, which MQTT 3.1.1 lacks. A v3.1.1 publish to `queries/` 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. The **immediate-PUBACK** rule applies exactly as in [Commands](/connectors/mqtt/how-to/commands#how-it-works): `PUBACK` acks receipt, not the response, so implement your own response-wait timeout on the `$reply` topic. ## How it works [#how-it-works] *The MQTT client is the caller; a gRPC responder answers the query and the response body returns on the client's `$reply` topic.* ## Request and response [#request-and-response] Each example subscribes to its own `$reply//inbox` topic, publishes a query to `queries/demo/rpc` with a response-topic and correlation-data, then reads the response **body** and `kubemq-metadata`. A gRPC responder must be running on channel `demo.rpc`. 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-query-" + uuid.NewString()[:8] correlationID := uuid.NewString() replyTopic := "$reply/" + clientID + "/inbox" // own namespace only queryTopic := "queries/demo/rpc" // channel demo.rpc 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) } if subAck.Reasons[0] == 0x83 { log.Fatal("SUBACK 0x83: reply topic must be $reply//...") } // 2. Publish the query with ResponseTopic + CorrelationData. pubAck, err := client.Publish(ctx, &paho.Publish{ Topic: queryTopic, QoS: 1, Payload: []byte(`{"action":"echo","data":"hello"}`), Properties: &paho.PublishProperties{ ResponseTopic: replyTopic, CorrelationData: []byte(correlationID), }, }) if err != nil { log.Fatalf("publish: %v", err) } // PUBACK is immediate — the response 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 — queries carry a body (plus kubemq-metadata). select { case r := <-resp: var metadata string if r.Properties != nil { for _, up := range r.Properties.User { if up.Key == "kubemq-metadata" { metadata = up.Value } } } fmt.Printf("query response body: %s (metadata=%s)\n", r.Payload, metadata) case <-ctx.Done(): log.Fatal("timed out — is a gRPC responder running on channel demo.rpc?") } _ = 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-query-client" QUERY_TOPIC = "queries/demo/rpc" # channel demo.rpc 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() result: dict = {} def on_message(client, userdata, msg): props = {} if msg.properties and hasattr(msg.properties, "UserProperty"): for k, v in (msg.properties.UserProperty or []): props[k] = v result["body"] = msg.payload result["metadata"] = props.get("kubemq-metadata") 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 query with ResponseTopic + CorrelationData. pub_props = Properties(PacketTypes.PUBLISH) pub_props.ResponseTopic = REPLY_TOPIC pub_props.CorrelationData = CORRELATION # PUBACK is immediate — the response arrives later on the reply topic. client.publish(QUERY_TOPIC, payload=b'{"query": "hello"}', qos=1, properties=pub_props).wait_for_publish(10) # 3. Wait for the response — queries carry a body (plus kubemq-metadata). if not response.wait(timeout=30): raise TimeoutError("no response — is a gRPC responder running on channel demo.rpc?") print(f"query response body: {result['body'].decode()!r} (metadata={result['metadata']})") client.loop_stop() client.disconnect() if __name__ == "__main__": main() ``` ```java import java.nio.charset.StandardCharsets; import java.util.UUID; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; 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-query-" + UUID.randomUUID().toString().substring(0, 6); String queryTopic = "queries/demo/rpc"; // channel demo.rpc String replyTopic = "$reply/" + clientId + "/inbox"; // own namespace only byte[] correlation = "corr-123".getBytes(StandardCharsets.UTF_8); BlockingQueue responses = new ArrayBlockingQueue<>(1); MqttAsyncClient client = new MqttAsyncClient(broker, clientId); client.setCallback(new MqttCallback() { public void messageArrived(String t, MqttMessage m) { responses.offer(m); } 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 query with ResponseTopic + CorrelationData. MqttProperties pubProps = new MqttProperties(); pubProps.setResponseTopic(replyTopic); pubProps.setCorrelationData(correlation); MqttMessage msg = new MqttMessage("ping".getBytes(StandardCharsets.UTF_8)); msg.setQos(1); msg.setProperties(pubProps); // PUBACK is immediate — the response arrives later on the reply topic. client.publish(queryTopic, msg).waitForCompletion(5_000); // 3. Wait for the response — queries carry a body (plus kubemq-metadata). MqttMessage response = responses.poll(15, TimeUnit.SECONDS); if (response == null) { throw new IllegalStateException("no response — is a gRPC responder running on channel demo.rpc?"); } String body = new String(response.getPayload(), StandardCharsets.UTF_8); String metadata = ""; MqttProperties props = response.getProperties(); if (props != null && props.getUserProperties() != null) { for (UserProperty up : props.getUserProperties()) { if ("kubemq-metadata".equals(up.getKey())) metadata = up.getValue(); } } System.out.printf("query response body: %s (metadata=%s)%n", body, metadata); client.disconnect().waitForCompletion(3_000); client.close(); } } ``` ```typescript import mqtt, { type MqttClient } from "mqtt"; import crypto from "node:crypto"; const CLIENT_ID = `js-query-${crypto.randomBytes(4).toString("hex")}`; const REPLY_TOPIC = `$reply/${CLIENT_ID}/inbox`; // own namespace only const QUERY_TOPIC = "queries/demo/rpc"; // channel demo.rpc 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); 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.rpc?")), 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 ?? {}; // Query responses carry a body (unlike commands). console.log(`query response body: ${payload.toString()} (metadata=${userProps["kubemq-metadata"] ?? ""})`); resolve(); }); }); await new Promise((resolve, reject) => { client.publish( QUERY_TOPIC, JSON.stringify({ request: "hello" }), { 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-query-{Guid.NewGuid():N}"[..23]; var requestId = Guid.NewGuid().ToString("N"); var replyTopic = $"$reply/{clientId}/inbox"; // own namespace only const string queryTopic = "queries/demo/rpc"; // channel demo.rpc var factory = new MqttFactory(); var responded = new TaskCompletionSource<(string body, string metadata)>( TaskCreationOptions.RunContinuationsAsynchronously); using var client = factory.CreateMqttClient(); client.ApplicationMessageReceivedAsync += e => { var body = Encoding.UTF8.GetString(e.ApplicationMessage.PayloadSegment); var metadata = string.Empty; foreach (var p in e.ApplicationMessage.UserProperties ?? new()) if (p.Name == "kubemq-metadata") metadata = p.Value; responded.TrySetResult((body, metadata)); 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 query with ResponseTopic + CorrelationData. var pubResult = await client.PublishAsync(new MqttApplicationMessageBuilder() .WithTopic(queryTopic) .WithPayload(Encoding.UTF8.GetBytes($"{{\"query\":\"hello\",\"requestId\":\"{requestId}\"}}")) .WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce) .WithResponseTopic(replyTopic) .WithCorrelationData(Encoding.UTF8.GetBytes(requestId)) .Build()); // PUBACK is immediate — the response arrives later on the reply topic. if (pubResult.ReasonCode != MqttClientPublishReasonCode.Success) throw new Exception($"PUBACK reason: {pubResult.ReasonCode}"); // 3. Wait for the response — queries carry a body (plus kubemq-metadata). using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15)); var (body, metadata) = await responded.Task.WaitAsync(cts.Token); Console.WriteLine($"query response body: {body} (metadata={metadata})"); 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-query-{}", &Uuid::new_v4().to_string()[..8]); let reply_topic = format!("$reply/{}/inbox", client_id); // own namespace only let query_topic = "queries/demo/rpc"; // channel demo.rpc 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::<(String, 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; } // Query responses carry a body (plus kubemq-metadata). let body = String::from_utf8_lossy(&p.payload).to_string(); let metadata = props .and_then(|pr| pr.user_properties.iter().find(|(k, _)| k == "kubemq-metadata")) .map(|(_, v)| v.clone()); if let Some(tx) = tx.take() { let _ = tx.send((body, metadata)); 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 query with ResponseTopic + CorrelationData. let props = PublishProperties { response_topic: Some(reply_topic.clone()), correlation_data: Some(corr_bytes), ..Default::default() }; // PUBACK is immediate — the response arrives later on the reply topic. client .publish_with_properties( query_topic, QoS::AtLeastOnce, false, br#"{"action":"echo","data":"hello"}"#.as_ref(), props, ) .await?; // 3. Wait for the response — queries carry a body (plus kubemq-metadata). let (body, metadata) = timeout(Duration::from_secs(30), rx) .await .map_err(|_| "timed out — is a gRPC responder running on channel demo.rpc?")??; println!("query response body: {body} (metadata={})", metadata.as_deref().unwrap_or("")); Ok(()) } ``` ## The response shape [#the-response-shape] A query response carries the responder's data: | Field | Type | Meaning | | --------------------- | ----------------- | --------------------------------------------------------------------- | | Payload | bytes | The response body returned by the gRPC responder | | `kubemq-metadata` | string (optional) | A metadata string set by the responder | | Other user properties | string | Any KubeMQ `Tags` the responder attached | | `CorrelationData` | bytes | Echoed from the request — use it to match the response to the request | Set a unique `CorrelationData` per request so you can match responses when multiple queries are in flight. The same reason codes and responder rules apply as for [Commands](/connectors/mqtt/how-to/commands#reason-codes-and-responders): `PUBACK 0x83` for a foreign `ResponseTopic`, `PUBACK 0x97` for `RpcMaxPending`, `SUBACK 0x83` for an MQTT client trying to subscribe as a responder. Responders run on the gRPC side (`SubscribeToQueries` + `SendQueryResponse`). ## Related [#related]