Responders

A responder reads the private reply address from reply-to and answers there through an injected slot, echoing correlation-id.

The reply address is new for every request, so neither the reply type nor the mount site can name it. The handler reads it with ctx.headers().reply_to() and publishes with .to(reply_to) at the call site. That call site exists only for a reply type whose Outgoing derive names no destination. The slot is bound as Out<impl Publisher> and mounted with .out(DefaultSlot, Publish).build().

The answer copies the request's correlation-id, which ctx.headers().correlation_id() reads. A requester matches the answer by that id, and an answer without it never resolves the request, which then times out. A message without reply-to is no request; the example drops it, which rejects the delivery.

A request whose payload is plain bytes reads them through a Deserialized type and answers with a Serialized one, so no codec touches either side.

Example

Macros
Manual

examples/amqp_request_reply.rs
/// The request payload: a name, not a serialized value, so it rides the byte lane and no codec
/// ever sees it.
#[derive(Deserialized)]
struct Who<'a>(&'a [u8]);

/// The reply the responder builds itself. It carries its own bytes and names no destination, so
/// the call site says where it goes, which is what a per-request reply address needs.
#[derive(Outgoing, Serialized)]
struct Greeting(Vec<u8>);

#[subscriber(AmqpAddress::queue("greeter"))]
async fn greet(
    who: &Who<'_>,
    ctx: &mut Context<'_>,
    Out(out): Out<impl Publisher>,
) -> HandlerOutcome {
    // The requester's private reply address; a message without one is no request.
    let Some(reply_to) = ctx.headers().reply_to().map(str::to_owned) else {
        return HandlerOutcome::drop();
    };
    let mut headers = HeaderMap::new();
    if let Some(correlation_id) = ctx.headers().correlation_id() {
        headers.insert("correlation-id", correlation_id.to_owned());
    }

    let name = String::from_utf8_lossy(who.0);
    tracing::info!(%name, %reply_to, "greeting");
    let greeting = Greeting(format!("hello, {name}").into_bytes());
    if out
        .message(&greeting)
        .to(reply_to)
        .with_headers(headers)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/amqp_request_reply.rs
/// The request payload: a name, not a serialized value, so it rides the byte lane and no codec
/// ever sees it.
#[derive(Deserialized)]
struct Who<'a>(&'a [u8]);

/// The reply the responder builds itself. It carries its own bytes and names no destination, so
/// the call site says where it goes, which is what a per-request reply address needs.
#[derive(Outgoing, Serialized)]
struct Greeting(Vec<u8>);

struct Greet;

impl<'p, Entry> Handle<Who<'p>, (), Outs<(Entry,)>> for Greet
where
    Entry: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        who: &Who<'p>,
        outs: &Outs<(Entry,)>,
        ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        // The requester's private reply address; a message without one is no request.
        let Some(reply_to) = ctx.headers().reply_to().map(str::to_owned) else {
            return Err(HandlerOutcome::drop());
        };
        let mut headers = HeaderMap::new();
        if let Some(correlation_id) = ctx.headers().correlation_id() {
            headers.insert("correlation-id", correlation_id.to_owned());
        }

        let name = String::from_utf8_lossy(who.0);
        tracing::info!(%name, %reply_to, "greeting");
        let greeting = Greeting(format!("hello, {name}").into_bytes());
        outs.get(DefaultSlot)
            .message(&greeting)
            .to(reply_to)
            .with_headers(headers)
            .publish()
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

Example​

Example