# Commands (/connectors/amqp/concepts/commands) Commands are **native, in-protocol request/reply** over the AMQP 1.0 connector. Attach a link to a node whose address begins with `commands/` and the connector binds it to the KubeMQ **Commands** pattern. RPC here is **fully native** — there is **no gRPC responder and no KubeMQ SDK**; a responder is just a normal AMQP consumer that publishes a reply. This page documents the shared RPC mechanics — the dynamic reply node, the anonymous responder, and correlation matching. [Queries](/connectors/amqp/concepts/queries) use the **same** request path and defer to this page; the only differences are the reply shape and the failure contract. ## Overview [#overview] A requester opens a dynamic reply node, sends a request to `commands/` naming that node as `reply-to`, and matches the reply by `correlation-id`. The connector fires `SendCommand`; the responder consumes `commands/`, does the work, and sends a reply back. | Step | Who | Action | | --------------- | --------- | -------------------------------------------------------------------------------------------- | | Open reply node | Requester | Attach a receiver with `source.dynamic = true`; the server mints `_amqp10.tmp..` | | Send request | Requester | `TRANSFER` to `commands/` with `reply-to` = the minted node + a `correlation-id` | | Route | Connector | Verifies reply-to ownership, then `SendCommand` | | Reply | Responder | Anonymous sender with `properties.to = /responses/` + the echoed `correlation-id` | | Match | Requester | Correlate the reply to its request by `correlation-id` | The reply's `correlation-id` is the request's `correlation-id`, or — when the request carried none — its **`message-id`** (the Qpid JMS convention). Set one or the other on every request and match the reply on it. ## How it works [#how-it-works] *The requester's dynamic reply node receives the responder's reply out-of-band; the requester matches it by `correlation-id`.* **`reply-to` must name a node this connection owns (snooping guard).** A requester cannot point `reply-to` at an arbitrary or foreign node — that would let it direct a response to another client's node. A **missing** `reply-to` returns `amqp:not-allowed` ("request missing reply-to"); a `reply-to` naming a node **this connection does not own** returns `amqp:not-allowed` ("reply-to is not a node this connection owns"). Always create a dynamic reply node per requester and use its echoed `_amqp10.tmp.*` address. ## Commands vs Queries — the failure contract [#commands-vs-queries--the-failure-contract] Both patterns share the dynamic-reply path, but their reply shape and failure behavior differ: | | **Commands** | **Queries** | | -------------------- | ----------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------- | | Reply body | optional | the result **body + metadata** | | Reply app-properties | `x-opt-kubemq-executed` (bool) **always**; `x-opt-kubemq-error` (string) when non-empty | **none** | | On success | `executed=true` | body + metadata returned | | **On failure** | a reply **is** delivered with **`executed=false`** (+ error text) — the requester is never left waiting | **nothing is delivered** — the requester **times out** (\~30 s) | In short: a **command** always answers (success or `executed=false`); a query answers on success and goes silent on failure. 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, so the snooping guard is honored), sends a successful command (`executed=true`) and a failing one (`executed=false`), and shows that **both** round-trip — neither hangs. Every client reads the broker endpoint from `KUBEMQ_AMQP_URL` (default `amqp://localhost:5672`). ```go 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/, 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/; 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) } ``` ```python 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() ``` ```java 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/, 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)); } } ``` ```csharp 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/, 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/; 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, }; ``` ```typescript 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/, reply via an anonymous sender. async function startResponder(address: string): Promise<{ stop: () => Promise }> { 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 { 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/; correlate replies. async function runRequester(address: string): Promise { 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 { 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; 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; } function awaitReply(replyReceiver: Receiver, timeoutMs: number): Promise { return new Promise((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 | undefined }); }; replyReceiver.on(ReceiverEvents.message, handler); }); } async function main(): Promise { 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); }); ``` ```rust 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) -> 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) -> Option { match id { Some(MessageId::String(s)) => Some(s.clone()), Some(other) => Some(format!("{other:?}")), None => None, } } fn command_outcome(msg: &Message>) -> (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> { 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)??; Ok(()) } // Responder: consume commands/, reply via an anonymous sender. async fn run_responder( url: &str, addr: &str, ready: oneshot::Sender<()>, mut stop: oneshot::Receiver<()>, ) -> Result<(), Box> { 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::>() => 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/; correlate replies. async fn run_requester(url: &str, addr: &str) -> Result<(), Box> { 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> { 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::>()).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(()) } ``` **`RpcMaxPending` bounds in-flight requests.** The default cap (512 per connection) limits outstanding requests; a request that cannot reserve a slot returns `amqp:resource-limit-exceeded` ("rpc pending limit reached"). Bound your concurrency accordingly. Dynamic reply nodes are node-local, but RPC **replies travel the broker reply path and are cluster-safe** — request/reply works across a cluster even though the reply node lives on the requester's node. ## Related [#related]