Skip to main content

Request-reply

This guide asks another service over NATS and waits for its answer, answers requests on their inbox, handles a request nobody answers, and scales the answering side.

NATS correlates a reply with its request itself. A request carries an inbox, a subject unique to it, and the answer is published to that inbox.

The example is two services in one app. The pricing service answers questions about prices, and the orders service asks it for the price of each order.

The responder​

src/main.rs
/// What the orders service asks the pricing service.
#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct Ask {
    sku: String,
}

/// The pricing service's answer, sent to the inbox of the request.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Price {
    sku: String,
    cents: u64,
}

/// The catalogue the pricing service answers from.
fn price_of(sku: &str) -> Option<u64> {
    match sku {
        "book" => Some(1250),
        "pen" => Some(150),
        _ => None,
    }
}

#[derive(OutSlot)]
#[publishes(Price)]
struct Answers;

#[subscriber(CoreSubject::new("prices.quote").queue_group("pricing"))]
async fn quote(
    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!(ask.sku, "a question without an inbox dropped");
        return HandlerOutcome::drop();
    };
    let Some(cents) = price_of(&ask.sku) else {
        tracing::info!(ask.sku, "no price for the item, no answer");
        return HandlerOutcome::drop();
    };
    let price = Price {
        sku: ask.sku.clone(),
        cents,
    };
    if answers.message(&price).to(inbox).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

A request arrives with its inbox in the reply-to header, and ctx.headers().reply_to() reads it. quote publishes the price through its Answers slot with to(inbox), because the destination is known only per request. Price therefore declares no subject of its own. An item without a price gets no answer, and the requester's timeout ends the wait.

The responder reads prices.quote in the queue group pricing. Each replica of the pricing service joins the group, and each request goes to one of them. Any NATS responder answers the same request, for example nats reply prices.quote '{"sku":"book","cents":1250}' from the command line.

The requester​

src/main.rs
/// How long the orders service waits for a price.
const PRICING_TIMEOUT: Duration = Duration::from_secs(2);

/// How long an order waits before it asks for its price again.
const RETRY_DELAY: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders.placed")]
struct Order {
    id: u64,
    sku: String,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders.priced")]
struct Priced {
    order: u64,
    cents: u64,
}

#[derive(OutSlot)]
struct Pricing;

#[subscriber(
    JetStreamSubject::new("orders.placed", "ORDERS").durable("pricer"),
    reply
)]
async fn price_order(
    order: &Order,
    Out(pricing): Out<impl RequestReply, Pricing>,
) -> Result<Priced, HandlerOutcome> {
    let Ok(ask) = serde_json::to_vec(&Ask {
        sku: order.sku.clone(),
    }) else {
        return Err(HandlerOutcome::drop());
    };
    let request = OutgoingMessage::new("prices.quote", ask.as_slice());
    let reply = match pricing.request(request, PRICING_TIMEOUT).await {
        Ok(reply) => reply,
        Err(err) => {
            tracing::warn!(order.id, %err, "no price, the order asks again later");
            return Err(HandlerOutcome::retry_after(RETRY_DELAY));
        }
    };
    let Ok(price) = serde_json::from_slice::<Price>(reply.payload()) else {
        return Err(HandlerOutcome::drop());
    };
    tracing::info!(order.id, price.cents, "order priced");
    Ok(Priced {
        order: order.id,
        cents: price.cents,
    })
}

price_order reads new orders from a stream and asks for each price on prices.quote. The Pricing slot is bounded Out<impl RequestReply, _>: the body says it needs request-reply and names no broker. The app pairs the slot with the plain Publish policy, whose publisher carries request-reply on NATS. request publishes the question with a fresh inbox and resolves with the first message on it. The question is a raw message, so the body encodes it and decodes the answer itself, here with serde_json. The order's reply, Priced, goes out once the price is known.

A request fails in two ways. When no subscription reads the subject, the server answers at once with "no responders". When a responder exists but stays silent, the request fails after its timeout, two seconds here, with NatsError::RequestTimeout. Either way the order waits 30 seconds on the stream and asks again.

The app​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(quote).out(Answers, Publish).build();
            b.include(price_order)
                .out_reply(Publish)
                .out(Pricing, Publish)
                .build();
        },
    )
}

Both slots take the plain Publish policy: one publishes answers, the other asks questions. A startup hook asks the same way, with b.after_startup(Publish, ..) and request on the publisher it is handed.

The tests in the example file run both halves in one app in process. The in-process NATS routes a request to the app's own responder, answers "no responders" when no subscription reads the subject, and times a silent responder out.