Queries

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)
	}
}

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()