# Queries (/connectors/amqp/concepts/queries)



Queries are **native, in-protocol request/reply** over the AMQP 1.0 connector, used to **fetch a value**. Attach a link to a node whose address begins with `queries/<channel>` and the connector binds it to the KubeMQ **Queries** pattern. A query returns a **result body + metadata**; on failure it delivers **nothing**, and the requester detects failure by **timeout**.

## Overview [#overview]

The request path is **identical** to [Commands](/connectors/amqp/concepts/commands): the requester opens a **dynamic reply node** (`source.dynamic = true`), sends to `queries/<ch>` with `reply-to` = the minted node + a `correlation-id`, and the responder replies via an **anonymous sender** to `properties.to = /responses/<RequestID>` carrying the echoed `correlation-id`. The same **snooping guard** (`reply-to` must name a connection-owned node) and the same correlation-id-with-message-id-fallback rule apply.

**See [Commands](/connectors/amqp/concepts/commands) for the full RPC mechanics** — dynamic reply nodes, the anonymous responder, the snooping guard, correlation matching, and the `RpcMaxPending` cap. This page covers only what differs for queries.

## How Queries differ from Commands [#how-queries-differ-from-commands]

|                      | **Commands**                                                 | **Queries**                                         |
| -------------------- | ------------------------------------------------------------ | --------------------------------------------------- |
| Reply body           | optional                                                     | the result **body + metadata**                      |
| Reply app-properties | `x-opt-kubemq-executed` + `x-opt-kubemq-error`               | **none**                                            |
| On success           | `executed=true`                                              | body + metadata returned                            |
| **On failure**       | a reply with `executed=false` (+ error) — never left waiting | **nothing delivered** — the requester **times out** |

A query is a "fetch a value" call: there is **no executed/error envelope**. When a query fails, times out, or the responder ignores it, the connector delivers no reply — so the requester's timeout **is** the failure signal. The connector's default per-request timeout is \~30 s; set the request's `header.ttl` (ms) to choose a per-request budget. Choose queries when a missing reply is an acceptable failure mode; choose commands when you need a positive failure signal.

## Request and reply [#request-and-reply]

Each example runs a responder and a requester (separate connections), sends a successful query (the reply round-trips with the result body) and a query the responder **ignores** (no reply, so the requester times out). Every client reads the broker endpoint from `KUBEMQ_AMQP_URL` (default `amqp://localhost:5672`).

<Tabs groupId="language" items="['Go','Python']">
  <Tab value="Go">
    ```go
    package main

    import (
    	"context"
    	"errors"
    	"fmt"
    	"log"
    	"os"
    	"sync"
    	"time"

    	amqp "github.com/Azure/go-amqp"
    )

    const channel = "amqp10.examples.queries"

    // A short per-request deadline so the "no reply" leg surfaces a timeout quickly.
    // The connector's own default RPC timeout is ~30s.
    const demoTimeout = 5 * time.Second

    func amqpURL() string {
    	if v := os.Getenv("KUBEMQ_AMQP_URL"); v != "" {
    		return v
    	}
    	return "amqp://localhost:5672"
    }

    func main() {
    	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
    	defer cancel()
    	addr := "queries/" + channel // queries/ prefix → KubeMQ Queries pattern

    	ready := make(chan struct{})
    	var wg sync.WaitGroup
    	wg.Add(1)
    	go func() { defer wg.Done(); runResponder(ctx, addr, ready) }()
    	<-ready

    	runRequester(ctx, addr)
    	cancel()
    	wg.Wait()
    }

    // Responder: a query whose body is "ignore" gets NO reply (the requester times out).
    func runResponder(ctx context.Context, addr string, ready chan<- struct{}) {
    	conn, _ := amqp.Dial(ctx, amqpURL(), nil)
    	defer func() { _ = conn.Close() }()
    	session, _ := conn.NewSession(ctx, nil)
    	rcv, _ := session.NewReceiver(ctx, addr, &amqp.ReceiverOptions{Credit: 10})
    	snd, _ := session.NewSender(ctx, "", nil) // anonymous reply sender (null target)
    	close(ready)

    	for {
    		req, err := rcv.Receive(ctx, nil)
    		if err != nil {
    			if ctx.Err() != nil || errors.Is(err, context.Canceled) {
    				return
    			}
    			return
    		}
    		_ = rcv.AcceptMessage(context.Background(), req)
    		if req.Properties == nil || req.Properties.ReplyTo == nil {
    			continue
    		}
    		body := string(req.GetData())
    		if body == "ignore" {
    			continue // send NOTHING — the requester times out
    		}

    		// A QUERY reply carries ONLY the body + metadata — no executed/error props.
    		replyTo := *req.Properties.ReplyTo
    		reply := amqp.NewMessage([]byte("result:" + body))
    		reply.Properties = &amqp.MessageProperties{To: &replyTo}
    		if req.Properties.CorrelationID != nil {
    			reply.Properties.CorrelationID = req.Properties.CorrelationID
    		} else {
    			reply.Properties.CorrelationID = req.Properties.MessageID
    		}
    		_ = snd.Send(ctx, reply, nil)
    	}
    }

    // Requester: dynamic reply node + sender on queries/<ch>; correlate replies.
    func runRequester(ctx context.Context, addr string) {
    	conn, _ := amqp.Dial(ctx, amqpURL(), nil)
    	defer func() { _ = conn.Close() }()
    	session, _ := conn.NewSession(ctx, nil)

    	// DYNAMIC reply node (see the Commands page for the shared mechanics).
    	replyRcv, _ := session.NewReceiver(ctx, "", &amqp.ReceiverOptions{DynamicAddress: true, Credit: 5})
    	replyNode := replyRcv.Address()
    	snd, _ := session.NewSender(ctx, addr, nil)

    	// 1. A SUCCESSFUL query: round-trips, body intact.
    	sendQuery(ctx, snd, replyNode, "get-temp-sensor-3", "corr-qry-1")
    	rcvCtx, c1 := context.WithTimeout(ctx, demoTimeout)
    	reply, err := replyRcv.Receive(rcvCtx, nil)
    	c1()
    	if err != nil {
    		log.Fatalf("await reply: %v", err)
    	}
    	_ = replyRcv.AcceptMessage(context.Background(), reply)
    	fmt.Printf("reply for %q: body=%q\n", "get-temp-sensor-3", string(reply.GetData()))

    	// 2. A query the responder ignores: NOTHING is delivered → the requester TIMES OUT.
    	//    The absence of a reply IS the failure signal for queries.
    	sendQuery(ctx, snd, replyNode, "ignore", "corr-qry-2")
    	rcvCtx2, c2 := context.WithTimeout(ctx, demoTimeout)
    	_, err = replyRcv.Receive(rcvCtx2, nil)
    	c2()
    	if err == nil {
    		log.Fatal("expected NO reply for \"ignore\"")
    	}
    	fmt.Printf("no reply for %q within %s — query timed out (expected)\n", "ignore", demoTimeout)
    }

    func sendQuery(ctx context.Context, snd *amqp.Sender, replyNode, body, corr string) {
    	req := amqp.NewMessage([]byte(body))
    	req.Properties = &amqp.MessageProperties{
    		ReplyTo:       &replyNode, // MUST name a node this connection owns (snooping guard)
    		CorrelationID: corr,
    	}
    	if err := snd.Send(ctx, req, nil); err != nil {
    		log.Fatalf("send query: %v", err)
    	}
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    import os
    import threading

    from proton import Message
    from proton.utils import BlockingConnection

    CHANNEL = "amqp10.examples.queries"
    # A short per-request deadline so the "no reply" leg surfaces a timeout quickly.
    # The connector's own default RPC timeout is ~30s.
    DEMO_TIMEOUT = 5.0


    def amqp_url() -> str:
        return os.environ.get("KUBEMQ_AMQP_URL", "amqp://localhost:5672")


    def run_responder(addr: str, ready: threading.Event, stop: threading.Event) -> None:
        # A query whose body is "ignore" gets NO reply (the requester times out).
        conn = BlockingConnection(amqp_url())
        try:
            rcv = conn.create_receiver(addr, credit=10)
            snd = conn.create_sender(None)  # anonymous reply sender (null target)
            ready.set()
            while not stop.is_set():
                try:
                    req = rcv.receive(timeout=1.0)
                except Exception:
                    continue
                rcv.accept()
                if not req.reply_to:
                    continue
                body = str(req.body)
                if body == "ignore":
                    continue  # send NOTHING — the requester times out

                # A QUERY reply carries ONLY the body + metadata — no executed/error props.
                reply = Message(body="result:" + body)
                reply.address = req.reply_to
                reply.correlation_id = req.correlation_id if req.correlation_id else req.id
                snd.send(reply)
        finally:
            conn.close()


    def run_requester(addr: str) -> None:
        conn = BlockingConnection(amqp_url())
        try:
            # DYNAMIC reply node (see the Commands page for the shared mechanics).
            reply_rcv = conn.create_receiver(None, dynamic=True, credit=5)
            reply_node = reply_rcv.link.remote_source.address
            snd = conn.create_sender(addr)

            # 1. A SUCCESSFUL query: round-trips, body intact.
            send_query(snd, reply_node, "get-temp-sensor-3", "corr-qry-1")
            reply = reply_rcv.receive(timeout=DEMO_TIMEOUT)
            reply_rcv.accept()
            print(f"reply for 'get-temp-sensor-3': body={str(reply.body)!r}")

            # 2. A query the responder ignores: NOTHING is delivered → the requester
            #    TIMES OUT. The absence of a reply IS the failure signal for queries.
            send_query(snd, reply_node, "ignore", "corr-qry-2")
            try:
                reply_rcv.receive(timeout=DEMO_TIMEOUT)
            except Exception:
                print(f"no reply for 'ignore' within {DEMO_TIMEOUT}s — query timed out (expected)")
            else:
                raise SystemExit("expected NO reply for 'ignore'")
        finally:
            conn.close()


    def send_query(snd, reply_node: str, body: str, corr: str) -> None:
        req = Message(body=body)
        req.reply_to = reply_node  # MUST name a node this connection owns (snooping guard)
        req.correlation_id = corr
        snd.send(req)


    def main() -> None:
        addr = "queries/" + CHANNEL  # queries/ prefix → KubeMQ Queries pattern
        ready, stop = threading.Event(), threading.Event()
        responder = threading.Thread(target=run_responder, args=(addr, ready, stop), daemon=True)
        responder.start()
        ready.wait(timeout=30.0)
        try:
            run_requester(addr)
        finally:
            stop.set()
            responder.join(timeout=10.0)


    if __name__ == "__main__":
        main()
    ```
  </Tab>
</Tabs>

<Callout type="info">
  The other languages (Java, C#, JavaScript, Rust) drive queries with the **same** code shape as [Commands](/connectors/amqp/concepts/commands) — the only changes are the `queries/` address prefix, dropping the `x-opt-kubemq-executed` / `x-opt-kubemq-error` application-properties on the reply, and treating a missing reply (timeout) as the failure signal. Adapt the Commands example for those languages.
</Callout>

## Related [#related]

<Cards>
  <Card title="Commands" href="/connectors/amqp/concepts/commands" description="The full shared RPC mechanics — dynamic reply nodes, anonymous responders, and correlation." />

  <Card title="Events" href="/connectors/amqp/concepts/events" description="Fire-and-forget pub/sub fan-out over AMQP 1.0." />

  <Card title="Address mapping" href="/connectors/amqp/reference/address-mapping" description="The address grammar, dynamic reply nodes, and the queries/ prefix." />
</Cards>