RequestReply

request publishes with a reply inbox and resolves with the first answer, or with NatsError::RequestTimeout when nothing answers in time.

NATS correlates an answer with its request itself, so the Core publisher, NatsPublisher, implements RequestReply. A handler bounds its slot Out<impl RequestReply, Slot>, and the mount site binds the plain Publish policy to it. The client subscribes to an inbox of its own and publishes the request with that inbox as the reply subject. The first message on the inbox is the answer, and the call returns it as the delivery it is; the handler decodes the payload. The request is bytes the handler encoded, and it goes to the connection directly, past the slot's transforms. When nothing subscribes to the subject, the server says so at once, and the call fails with NatsError::Publish instead of waiting out the timeout. A request is Core NATS: it reaches a responder listening at that moment, and nothing stores it. The same slot also publishes with the builder, what its #[publishes(..)] list admits.

Example

The handler asks the price service for the price of an order's item, then publishes the priced order through the same slot.

Macros
Manual

examples/nats_request_reply.rs
#[derive(OutSlot)]
#[publishes(Priced)]
struct Pricing;

/// Asks the price service for the price of the order's item, then publishes the priced order
/// through the same slot. No answer within two seconds puts the order back in its stream.
#[subscriber(JetStreamSubject::new("orders.placed", "ORDERS").durable("pricer"))]
async fn price(order: &Order, Out(pricing): Out<impl RequestReply, Pricing>) -> HandlerOutcome {
    let Ok(ask) = serde_json::to_vec(&Ask {
        item: order.item.clone(),
    }) else {
        return HandlerOutcome::drop();
    };
    let request = OutgoingMessage::new("prices.quote", ask.as_slice());
    let answer = match pricing.request(request, Duration::from_secs(2)).await {
        Ok(answer) => answer,
        Err(err) => {
            tracing::warn!(order.id, %err, "no price, redelivering");
            return HandlerOutcome::retry();
        }
    };
    let Ok(quote) = serde_json::from_slice::<Quote>(answer.payload()) else {
        return HandlerOutcome::drop();
    };
    let priced = Priced {
        id: order.id,
        cents: quote.cents,
    };
    if pricing.message(&priced).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/nats_request_reply.rs
#[derive(OutSlot)]
#[publishes(Priced)]
struct Pricing;

/// Asks the price service for the price of the order's item, then publishes the priced order
/// through the same slot. No answer within two seconds puts the order back in its stream.
struct Price;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Price
where
    Entry: OutEntry<Pricing, Wire: RequestReply>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let pricing = outs.get(Pricing);
        let ask = serde_json::to_vec(&Ask {
            item: order.item.clone(),
        })
        .map_err(|_| HandlerOutcome::drop())?;
        let request = OutgoingMessage::new("prices.quote", ask.as_slice());
        let answer = pricing
            .request(request, Duration::from_secs(2))
            .await
            .map_err(|err| {
                tracing::warn!(order.id, %err, "no price, redelivering");
                HandlerOutcome::retry()
            })?;
        let quote = serde_json::from_slice::<Quote>(answer.payload())
            .map_err(|_| HandlerOutcome::drop())?;
        let priced = Priced {
            id: order.id,
            cents: quote.cents,
        };
        pricing
            .message(&priced)
            .publish()
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

The mount binds Publish to the slot; the price service runs in the same app here.

Macros
Manual

examples/nats_request_reply.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            // The plain policy pairs into the Core publisher, which also serves requests.
            b.include(price).out(Pricing, Publish).build();
            b.include(quote_price).out(Answers, Publish).build();
        },
    )
}

examples/manual/nats_request_reply.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            let placed = JetStreamSubject::new("orders.placed", "ORDERS").durable("pricer");
            // The plain policy pairs into the Core publisher, which also serves requests.
            b.include(subscriber(placed, Price).build())
                .out(Pricing, Publish)
                .build();
            b.include(subscriber("prices.quote", QuotePrice).build())
                .out(Answers, Publish)
                .build();
        },
    )
}

Example​

Example