Skip to main content

Responders

A responder reads the request's inbox from the reply-to header and answers there through an injected publisher.

NATS carries a request's inbox in the protocol's reply field, and this crate surfaces it on a Core delivery as the reply-to header. The handler reads it with ctx.headers().reply_to(). The answer leaves through an Out slot with .to(inbox), so the answer type declares no subject of its own. The answer is an ordinary Core publish to the inbox, and the requester's client matches it to the waiting request. A message without an inbox was not a request, and a redelivery would not add one, so the handler drops it. A responder reads a CoreSubject, the bare name form, or a CoreWildcard. On a JetStream delivery the reply field is the consumer's acknowledgement subject, so no reply-to header is surfaced there. Any NATS client's request reaches the responder, nats request included, and a RequestReply slot of another service is one of them. With a queue group on the subscription, each request is answered by one member of the group.

Example​

examples/nats_request_reply_responder.rs
/// The slot a quote leaves through. A quote declares no subject: each request names its own.
#[derive(OutSlot)]
#[publishes(Quote)]
struct Answers;

/// Answers each request on the inbox it named. A message without an inbox was not a request,
/// and a redelivery would not grow one, so it is dropped.
#[subscriber("prices.quote")]
async fn quote_price(
    ask: &Ask,
    ctx: &mut Context<'_>,
    Out(answers): Out<impl Publisher, Answers>,
) -> HandlerOutcome {
    let Some(inbox) = ctx.headers().reply_to().map(str::to_owned) else {
        tracing::warn!(item = %ask.item, "a quote asked without an inbox");
        return HandlerOutcome::drop();
    };
    let quote = Quote {
        item: ask.item.clone(),
        cents: 250,
    };
    if answers.message(&quote).to(inbox).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}