Commands

package main

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

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

const channel = "amqp10.examples.commands"

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 := "commands/" + channel // commands/ prefix → KubeMQ Commands 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: consume commands/<ch>, reply via an anonymous sender.
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 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())

		// A command body of "fail" is rejected (executed=false); both paths reply.
		ok := body != "fail"
		errText := ""
		if !ok {
			errText = "command rejected by handler"
		}
		replyTo := *req.Properties.ReplyTo
		reply := amqp.NewMessage([]byte("ack:" + 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
		}
		// A COMMAND reply carries the execution outcome as application-properties.
		reply.ApplicationProperties = map[string]any{
			"x-opt-kubemq-executed": ok,
			"x-opt-kubemq-error":    errText,
		}
		_ = snd.Send(ctx, reply, nil)
	}
}

// Requester: dynamic reply node + sender on commands/<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: empty source + DynamicAddress:true → server echoes its address.
	replyRcv, _ := session.NewReceiver(ctx, "", &amqp.ReceiverOptions{DynamicAddress: true, Credit: 5})
	replyNode := replyRcv.Address()
	snd, _ := session.NewSender(ctx, addr, nil)

	doRequest(ctx, snd, replyRcv, replyNode, "reboot-node-7", "corr-cmd-1") // executed=true
	doRequest(ctx, snd, replyRcv, replyNode, "fail", "corr-cmd-2")          // executed=false
}

func doRequest(ctx context.Context, snd *amqp.Sender, replyRcv *amqp.Receiver, 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 command: %v", err)
	}
	// A command ALWAYS replies (success or failure), so this never times out.
	reply, err := replyRcv.Receive(ctx, nil)
	if err != nil {
		log.Fatalf("await reply: %v", err)
	}
	_ = replyRcv.AcceptMessage(context.Background(), reply)
	executed, _ := reply.ApplicationProperties["x-opt-kubemq-executed"].(bool)
	errText, _ := reply.ApplicationProperties["x-opt-kubemq-error"].(string)
	fmt.Printf("reply for %q: executed=%v error=%q\n", body, executed, errText)
}

import os
import threading

from proton import Message
from proton.utils import BlockingConnection

CHANNEL = "amqp10.examples.commands"
EXECUTED_PROP = "x-opt-kubemq-executed"
ERROR_PROP = "x-opt-kubemq-error"


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

            # A command body of "fail" is rejected (executed=false); both paths reply.
            ok = body != "fail"
            err_text = "" if ok else "command rejected by handler"
            reply = Message(body="ack:" + body)
            reply.address = req.reply_to
            reply.correlation_id = req.correlation_id if req.correlation_id else req.id
            # A COMMAND reply carries the execution outcome as application-properties.
            reply.properties = {EXECUTED_PROP: ok, ERROR_PROP: err_text}
            snd.send(reply)
    finally:
        conn.close()


def do_request(snd, reply_rcv, 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)
    # A command ALWAYS replies (success or failure), so this never times out.
    reply = reply_rcv.receive(timeout=30.0)
    reply_rcv.accept()
    props = reply.properties or {}
    print(f"reply for {body!r}: executed={bool(props.get(EXECUTED_PROP))} error={str(props.get(ERROR_PROP, ''))!r}")


def run_requester(addr: str) -> None:
    conn = BlockingConnection(amqp_url())
    try:
        # DYNAMIC reply node: dynamic=True → server echoes its address.
        reply_rcv = conn.create_receiver(None, dynamic=True, credit=5)
        reply_node = reply_rcv.link.remote_source.address
        snd = conn.create_sender(addr)
        do_request(snd, reply_rcv, reply_node, "reboot-node-7", "corr-cmd-1")  # executed=true
        do_request(snd, reply_rcv, reply_node, "fail", "corr-cmd-2")           # executed=false
    finally:
        conn.close()


def main() -> None:
    addr = "commands/" + CHANNEL  # commands/ prefix → KubeMQ Commands 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()

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TextMessage;

import org.apache.qpid.jms.JmsConnectionFactory;

public final class Main {
    private static final String CHANNEL = "amqp10.examples.commands";
    private static final String PROP_EXECUTED = "x-opt-kubemq-executed";
    private static final String PROP_ERROR = "x-opt-kubemq-error";

    public static void main(String[] args) throws Exception {
        String url = System.getenv().getOrDefault("KUBEMQ_AMQP_URL", "amqp://localhost:5672");
        String address = "commands/" + CHANNEL; // commands/ prefix → KubeMQ Commands

        // jms.validatePropertyNames=false lets us set/read the hyphenated
        // x-opt-kubemq-executed / -error application-properties.
        JmsConnectionFactory factory = new JmsConnectionFactory(url);
        factory.setValidatePropertyNames(false);

        Thread responder = new Thread(() -> runResponder(factory, address), "responder");
        responder.setDaemon(true);
        responder.start();
        Thread.sleep(1_000); // let the responder attach before sending

        runRequester(factory, address);
    }

    // Responder: consume commands/<ch>, reply to each request's JMSReplyTo.
    private static void runResponder(JmsConnectionFactory factory, String address) {
        try (Connection connection = factory.createConnection()) {
            connection.start();
            try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
                Queue commands = session.createQueue(address);
                try (MessageConsumer consumer = session.createConsumer(commands);
                        MessageProducer replyProducer = session.createProducer(null)) { // unidentified
                    while (true) {
                        Message req = consumer.receive(1_000);
                        if (req == null) continue;
                        Destination replyTo = req.getJMSReplyTo();
                        if (replyTo == null) continue;
                        String body = (req instanceof TextMessage) ? ((TextMessage) req).getText() : "";

                        boolean ok = !"fail".equals(body);
                        String errText = ok ? "" : "command rejected by handler";
                        TextMessage reply = session.createTextMessage("ack:" + body);
                        reply.setJMSCorrelationID(req.getJMSCorrelationID());
                        reply.setBooleanProperty(PROP_EXECUTED, ok);
                        reply.setStringProperty(PROP_ERROR, errText);
                        replyProducer.send(replyTo, reply);
                    }
                }
            }
        } catch (Exception e) {
            // connection torn down on shutdown
        }
    }

    // Requester: a JMS temporary queue is the dynamic reply node.
    private static void runRequester(JmsConnectionFactory factory, String address) throws Exception {
        try (Connection connection = factory.createConnection()) {
            connection.start();
            try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
                TemporaryQueue replyNode = session.createTemporaryQueue();
                Queue commands = session.createQueue(address);
                try (MessageConsumer replyConsumer = session.createConsumer(replyNode);
                        MessageProducer producer = session.createProducer(commands)) {
                    doRequest(session, producer, replyConsumer, replyNode, "reboot-node-7", "corr-cmd-1");
                    doRequest(session, producer, replyConsumer, replyNode, "fail", "corr-cmd-2");
                }
            }
        }
    }

    private static void doRequest(Session session, MessageProducer producer, MessageConsumer replyConsumer,
            TemporaryQueue replyNode, String body, String corr) throws Exception {
        TextMessage req = session.createTextMessage(body);
        req.setJMSReplyTo(replyNode); // MUST name a node this connection owns (snooping guard)
        req.setJMSCorrelationID(corr);
        producer.send(req);
        Message reply = replyConsumer.receive(30_000); // a command always replies
        if (reply == null) throw new IllegalStateException("timed out awaiting reply");
        System.out.printf("reply for \"%s\": executed=%b error=\"%s\"%n",
                body, reply.getBooleanProperty(PROP_EXECUTED), reply.getStringProperty(PROP_ERROR));
    }
}

using System.Text;
using Amqp;
using Amqp.Framing;

const string channel = "amqp10.examples.commands";

static string AmqpUrl() =>
    Environment.GetEnvironmentVariable("KUBEMQ_AMQP_URL") is { Length: > 0 } v
        ? v
        : "amqp://localhost:5672";

var addr = "commands/" + channel; // commands/ prefix → KubeMQ Commands pattern

using var responderDone = new CancellationTokenSource();
var responderReady = new TaskCompletionSource();
var responderTask = Task.Run(() => RunResponder(addr, responderReady, responderDone.Token));
await responderReady.Task.WaitAsync(TimeSpan.FromSeconds(20));

await RunRequester(addr);
responderDone.Cancel();
await responderTask;

// Responder: consume commands/<ch>, reply via an anonymous sender.
static void RunResponder(string addr, TaskCompletionSource ready, CancellationToken stop)
{
    var connection = Connection.Factory.CreateAsync(new Address(AmqpUrl())).GetAwaiter().GetResult();
    try
    {
        var session = new Session(connection);
        var receiver = new ReceiverLink(session, "command-responder", addr);
        receiver.SetCredit(10, autoRestore: true);
        var anonAttach = new Attach { Source = new Source(), Target = null }; // anonymous sender
        var sender = new SenderLink(session, "command-reply-sender", anonAttach, null);
        ready.TrySetResult();

        while (!stop.IsCancellationRequested)
        {
            var req = receiver.Receive(TimeSpan.FromSeconds(1));
            if (req is null) continue;
            receiver.Accept(req);
            if (req.Properties?.ReplyTo is not { Length: > 0 } replyTo) continue;
            var body = BodyString(req);

            var ok = body != "fail";
            var errText = ok ? "" : "command rejected by handler";
            var reply = new Message { BodySection = new Data { Binary = Encoding.UTF8.GetBytes("ack:" + body) } };
            reply.Properties = new Properties { To = replyTo };
            var corr = req.Properties.GetCorrelationId() ?? req.Properties.GetMessageId();
            if (corr is not null) reply.Properties.SetCorrelationId(corr);
            // A COMMAND reply carries the execution outcome as application-properties.
            reply.ApplicationProperties = new ApplicationProperties();
            reply.ApplicationProperties.Map["x-opt-kubemq-executed"] = ok;
            reply.ApplicationProperties.Map["x-opt-kubemq-error"] = errText;
            sender.Send(reply, TimeSpan.FromSeconds(10));
        }
    }
    catch (Exception ex) when (stop.IsCancellationRequested) { _ = ex; }
    finally { try { connection.CloseAsync().Wait(2000); } catch (AmqpException) { } }
}

// Requester: dynamic reply node + sender on commands/<ch>; correlate replies.
static async Task RunRequester(string addr)
{
    var connection = await Connection.Factory.CreateAsync(new Address(AmqpUrl()));
    try
    {
        var session = new Session(connection);

        // DYNAMIC reply node: Source.Dynamic = true; the server echoes the address
        // via the OnAttached callback.
        string? replyNode = null;
        using var attached = new SemaphoreSlim(0, 1);
        var dynAttach = new Attach { Source = new Source { Dynamic = true }, Target = new Target() };
        var replyRcv = new ReceiverLink(session, "command-reply-node", dynAttach, (link, attach) =>
        {
            if (attach.Source is Source s) replyNode = s.Address;
            attached.Release();
        });
        replyRcv.SetCredit(5, autoRestore: true);
        await attached.WaitAsync(TimeSpan.FromSeconds(10));

        var sender = new SenderLink(session, "command-requester", addr);
        DoRequest(sender, replyRcv, replyNode!, "reboot-node-7", "corr-cmd-1"); // executed=true
        DoRequest(sender, replyRcv, replyNode!, "fail", "corr-cmd-2");          // executed=false

        await sender.CloseAsync();
        await replyRcv.CloseAsync();
        await session.CloseAsync();
    }
    finally { await connection.CloseAsync(); }
}

static void DoRequest(SenderLink sender, ReceiverLink replyRcv, string replyNode, string body, string corr)
{
    var req = new Message { BodySection = new Data { Binary = Encoding.UTF8.GetBytes(body) } };
    req.Properties = new Properties { ReplyTo = replyNode }; // snooping guard: must own this node
    req.Properties.SetCorrelationId(corr);
    sender.Send(req, TimeSpan.FromSeconds(15));
    var reply = replyRcv.Receive(TimeSpan.FromSeconds(30)) // a command always replies
        ?? throw new InvalidOperationException("await reply: timed out");
    replyRcv.Accept(reply);
    var executed = reply.ApplicationProperties?.Map["x-opt-kubemq-executed"] as bool? ?? false;
    var errText = reply.ApplicationProperties?.Map["x-opt-kubemq-error"] as string ?? "";
    Console.WriteLine($"reply for \"{body}\": executed={executed} error=\"{errText}\"");
}

static string BodyString(Message message) => message.BodySection switch
{
    Data d => Encoding.UTF8.GetString(d.Binary),
    AmqpValue { Value: byte[] bytes } => Encoding.UTF8.GetString(bytes),
    AmqpValue { Value: string str } => str,
    AmqpValue v => v.Value?.ToString() ?? string.Empty,
    _ => string.Empty,
};

import {
  Connection,
  ReceiverEvents,
  type AwaitableSender,
  type ConnectionOptions,
  type EventContext,
  type Receiver,
} from "rhea-promise";

const channel = "amqp10.examples.commands";

function connectionOptions(suffix: string): ConnectionOptions {
  const url = new URL(process.env["KUBEMQ_AMQP_URL"] ?? "amqp://localhost:5672");
  return {
    host: url.hostname,
    port: url.port ? Number(url.port) : 5672,
    container_id: `kubemq-amqp10-js-commands-${suffix}-${process.pid}`,
    reconnect: false,
  };
}

function bodyToString(body: unknown): string {
  if (Buffer.isBuffer(body)) return body.toString("utf8");
  if (typeof body === "string") return body;
  return "";
}

// Responder: consume commands/<ch>, reply via an anonymous sender.
async function startResponder(address: string): Promise<{ stop: () => Promise<void> }> {
  const connection = new Connection(connectionOptions("responder"));
  await connection.open();
  const receiver = await connection.createReceiver({
    source: { address }, credit_window: 0, autoaccept: false, autosettle: false,
  });
  const replySender = await connection.createAwaitableSender({ target: {} }); // null target

  receiver.on(ReceiverEvents.message, (ctx: EventContext) => {
    void onRequest(replySender, ctx).catch((err) => console.error(err));
  });
  receiver.addCredit(10);

  return {
    stop: async () => {
      await replySender.close();
      await receiver.close();
      await connection.close();
    },
  };
}

async function onRequest(replySender: AwaitableSender, ctx: EventContext): Promise<void> {
  const req = ctx.message;
  ctx.delivery?.accept();
  if (!req || req.reply_to === undefined || req.reply_to === null) return;
  const body = bodyToString(req.body);

  const ok = body !== "fail";
  const errText = ok ? "" : "command rejected by handler";
  await replySender.send(
    {
      body: `ack:${body}`,
      to: req.reply_to,
      correlation_id: req.correlation_id ?? req.message_id,
      // A COMMAND reply carries the execution outcome as application-properties.
      application_properties: { "x-opt-kubemq-executed": ok, "x-opt-kubemq-error": errText },
    },
    { timeoutInSeconds: 10 },
  );
}

// Requester: dynamic reply node + sender on commands/<ch>; correlate replies.
async function runRequester(address: string): Promise<void> {
  const connection = new Connection(connectionOptions("requester"));
  await connection.open();
  try {
    const replyReceiver = await connection.createReceiver({
      source: { address: "", dynamic: true }, // server names the node
      credit_window: 0, autoaccept: false, autosettle: false,
    });
    replyReceiver.addCredit(5);
    const replyNode = replyReceiver.address || replyReceiver.source.address;
    const sender = await connection.createAwaitableSender({ target: { address } });

    await doRequest(sender, replyReceiver, replyNode!, "reboot-node-7", "corr-cmd-1"); // executed=true
    await doRequest(sender, replyReceiver, replyNode!, "fail", "corr-cmd-2");          // executed=false

    await sender.close();
    await replyReceiver.close();
  } finally {
    await connection.close();
  }
}

async function doRequest(
  sender: AwaitableSender, replyReceiver: Receiver, replyNode: string, body: string, corr: string,
): Promise<void> {
  const replyPromise = awaitReply(replyReceiver, 30_000); // arm before sending
  await sender.send(
    { body, reply_to: replyNode, correlation_id: corr }, // snooping guard: must own replyNode
    { timeoutInSeconds: 15 },
  );
  const reply = await replyPromise; // a command always replies
  const props = (reply.application_properties ?? {}) as Record<string, unknown>;
  const executed = props["x-opt-kubemq-executed"] === true;
  const errText = typeof props["x-opt-kubemq-error"] === "string" ? props["x-opt-kubemq-error"] : "";
  console.log(`reply for "${body}": executed=${executed} error="${errText}"`);
}

interface ReplyMessage {
  application_properties?: Record<string, unknown>;
}

function awaitReply(replyReceiver: Receiver, timeoutMs: number): Promise<ReplyMessage> {
  return new Promise<ReplyMessage>((resolve, reject) => {
    const timer = setTimeout(() => {
      replyReceiver.removeListener(ReceiverEvents.message, handler);
      reject(new Error("timed out awaiting reply"));
    }, timeoutMs);
    const handler = (ctx: EventContext): void => {
      clearTimeout(timer);
      replyReceiver.removeListener(ReceiverEvents.message, handler);
      ctx.delivery?.accept();
      replyReceiver.addCredit(1);
      resolve({ application_properties: ctx.message?.application_properties as Record<string, unknown> | undefined });
    };
    replyReceiver.on(ReceiverEvents.message, handler);
  });
}

async function main(): Promise<void> {
  const address = `commands/${channel}`; // commands/ prefix → KubeMQ Commands pattern
  const responder = await startResponder(address);
  try {
    await runRequester(address);
  } finally {
    await responder.stop();
  }
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

use std::time::Duration;

use fe2o3_amqp::link::receiver::CreditMode;
use fe2o3_amqp::{Connection, Receiver, Sender, Session};
use fe2o3_amqp_types::definitions::SenderSettleMode;
use fe2o3_amqp_types::messaging::{
    ApplicationProperties, Body, Message, MessageId, Properties, Source, Target,
};
use fe2o3_amqp_types::primitives::{SimpleValue, Value};
use tokio::sync::oneshot;

const CHANNEL: &str = "amqp10.examples.commands";
const EXECUTED_PROP: &str = "x-opt-kubemq-executed";
const ERROR_PROP: &str = "x-opt-kubemq-error";

fn amqp_url() -> String {
    std::env::var("KUBEMQ_AMQP_URL").unwrap_or_else(|_| "amqp://localhost:5672".to_string())
}

fn body_string(body: &Body<Value>) -> String {
    let bytes = match body {
        Body::Data(batch) => batch.iter().flat_map(|d| d.0.to_vec()).collect(),
        Body::Value(v) => match &v.0 {
            Value::Binary(b) => b.to_vec(),
            Value::String(s) => s.clone().into_bytes(),
            other => format!("{other:?}").into_bytes(),
        },
        _ => Vec::new(),
    };
    String::from_utf8_lossy(&bytes).into_owned()
}

fn message_id_string(id: &Option<MessageId>) -> Option<String> {
    match id {
        Some(MessageId::String(s)) => Some(s.clone()),
        Some(other) => Some(format!("{other:?}")),
        None => None,
    }
}

fn command_outcome(msg: &Message<Body<Value>>) -> (bool, String) {
    let Some(props) = &msg.application_properties else {
        return (false, String::new());
    };
    let executed = matches!(props.get(EXECUTED_PROP), Some(SimpleValue::Bool(true)));
    let error = match props.get(ERROR_PROP) {
        Some(SimpleValue::String(s)) => s.clone(),
        _ => String::new(),
    };
    (executed, error)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let url = amqp_url();
    let addr = format!("commands/{CHANNEL}"); // commands/ prefix → KubeMQ Commands

    let (ready_tx, ready_rx) = oneshot::channel::<()>();
    let (stop_tx, stop_rx) = oneshot::channel::<()>();
    let (rurl, raddr) = (url.clone(), addr.clone());
    let responder = tokio::spawn(async move { run_responder(&rurl, &raddr, ready_tx, stop_rx).await });
    ready_rx.await.map_err(|_| "responder failed to become ready")?;

    run_requester(&url, &addr).await?;
    let _ = stop_tx.send(());
    responder.await.map_err(|e| Box::new(e) as Box<dyn std::error::Error + Send + Sync>)??;
    Ok(())
}

// Responder: consume commands/<ch>, reply via an anonymous sender.
async fn run_responder(
    url: &str, addr: &str, ready: oneshot::Sender<()>, mut stop: oneshot::Receiver<()>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let mut connection = Connection::open("amqp10-examples-commands-responder", url).await?;
    let mut session = Session::begin(&mut connection).await?;
    let mut rcv = Receiver::builder()
        .name("commands-responder-receiver")
        .source(addr)
        .credit_mode(CreditMode::Auto(10))
        .attach(&mut session)
        .await?;
    // Anonymous sender (null target). Senders must set an explicit settle-mode
    // (the connector rejects the AMQP default `mixed`).
    let mut snd = Sender::builder()
        .name("commands-responder-anon-sender")
        .target(Target::builder().build())
        .sender_settle_mode(SenderSettleMode::Unsettled)
        .attach(&mut session)
        .await?;
    let _ = ready.send(());

    loop {
        let delivery = tokio::select! {
            biased;
            _ = &mut stop => break,
            res = rcv.recv::<Body<Value>>() => match res { Ok(d) => d, Err(_) => break },
        };
        rcv.accept(&delivery).await?;
        let msg = delivery.message();
        let Some(reply_to) = msg.properties.as_ref().and_then(|p| p.reply_to.clone()) else {
            continue;
        };
        let body = body_string(&msg.body);

        let ok = body != "fail";
        let err_text = if ok { String::new() } else { "command rejected by handler".to_string() };
        let corr = msg.properties.as_ref()
            .and_then(|p| p.correlation_id.clone().or_else(|| p.message_id.clone()));
        let mut props = Properties::builder().to(reply_to);
        if let Some(c) = corr {
            props = props.correlation_id(c);
        }
        let reply = Message::builder()
            .properties(props.build())
            .application_properties(
                ApplicationProperties::builder()
                    .insert(EXECUTED_PROP, ok)
                    .insert(ERROR_PROP, err_text.as_str())
                    .build(),
            )
            .data(format!("ack:{body}").into_bytes())
            .build();
        snd.send(reply).await?;
    }
    snd.close().await?;
    rcv.close().await?;
    session.end().await?;
    connection.close().await?;
    Ok(())
}

// Requester: dynamic reply node + sender on commands/<ch>; correlate replies.
async fn run_requester(url: &str, addr: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let mut connection = Connection::open("amqp10-examples-commands-requester", url).await?;
    let mut session = Session::begin(&mut connection).await?;

    // DYNAMIC reply node: empty source + dynamic(true); the server echoes the address.
    let mut reply_rcv = Receiver::builder()
        .name("commands-requester-reply-node")
        .source(Source::builder().dynamic(true).build())
        .credit_mode(CreditMode::Auto(5))
        .attach(&mut session)
        .await?;
    let reply_node = reply_rcv.source().as_ref()
        .and_then(|s| s.address.clone())
        .ok_or("server did not assign a dynamic reply-node address")?;
    let mut snd = Sender::builder()
        .name("commands-requester-sender")
        .target(addr)
        .sender_settle_mode(SenderSettleMode::Unsettled)
        .attach(&mut session)
        .await?;

    do_request(&mut snd, &mut reply_rcv, &reply_node, "reboot-node-7", "corr-cmd-1").await?; // executed=true
    do_request(&mut snd, &mut reply_rcv, &reply_node, "fail", "corr-cmd-2").await?;          // executed=false

    snd.close().await?;
    reply_rcv.close().await?;
    session.end().await?;
    connection.close().await?;
    Ok(())
}

async fn do_request(
    snd: &mut Sender, reply_rcv: &mut Receiver, reply_node: &str, body: &str, corr: &str,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let req = Message::builder()
        .properties(
            Properties::builder()
                .reply_to(reply_node.to_string()) // snooping guard: must own this node
                .correlation_id(corr.to_string())
                .build(),
        )
        .data(body.as_bytes().to_vec())
        .build();
    let outcome = snd.send(req).await?;
    if !outcome.is_accepted() {
        return Err(format!("send command {body:?}: unexpected outcome {outcome:?}").into());
    }
    // A command ALWAYS replies (success or failure), so this never times out.
    let reply = match tokio::time::timeout(Duration::from_secs(30), reply_rcv.recv::<Body<Value>>()).await {
        Ok(Ok(r)) => r,
        _ => return Err(format!("await reply for {body:?}: timed out").into()),
    };
    reply_rcv.accept(&reply).await?;
    let (executed, err_text) = command_outcome(reply.message());
    let _ = message_id_string(&reply.message().properties.as_ref().and_then(|p| p.correlation_id.clone()));
    println!("reply for {body:?}: executed={executed} error={err_text:?}");
    Ok(())
}