RequestReply

RequestReply sends a request from a handler and awaits the correlated answer, on a broker that correlates replies natively.

NATS-style brokers correlate replies, and the memory broker does too; Kafka and classic queues do not offer this capability. A handler bounds its slot Out<impl RequestReply, Marker>, and the mount site binds the broker's request policy, which its prelude names Request. request(message, timeout) publishes the message and resolves on the first answer, or fails once the time runs out. The request is an OutgoingMessage of bytes the handler encoded, and the answer is the broker's incoming message, which the handler decodes. The request goes to the broker directly, past the slot's transforms and the app's publish layers. The broker names an inbox for the answer: the memory broker puts it in the reply-to header. The responder is an ordinary reply handler whose reply goes to that inbox, through a naming transform. The slot also publishes with the builder, what its #[publishes(..)] list admits.

Example

The handler asks the quote service for a price, then publishes the priced order through the same slot.

Macros

examples/requester.rs

#[derive(OutSlot)]
#[publishes(Priced)]
struct Pricing;

#[subscriber("orders")]
async fn price(order: &Order, Out(pricing): Out<impl RequestReply, Pricing>) -> HandlerOutcome {
    let Ok(ask) = serde_json::to_vec(&Ask { id: order.id }) else {
        return HandlerOutcome::drop();
    };
    let request = OutgoingMessage::new("quotes.ask", ask.as_slice());
    let Ok(answer) = pricing.request(request, Duration::from_secs(2)).await else {
        return HandlerOutcome::retry();
    };
    let Ok(quote) = serde_json::from_slice::<Quote>(answer.payload()) else {
        return HandlerOutcome::drop();
    };
    let priced = Priced {
        id: order.id,
        price: quote.price,
    };
    if pricing.message(&priced).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/requester.rs

struct Pricing;

impl OutSlot for Pricing {
    const NAME: &'static str = "Pricing";
    type Destination = Reads;
}

impl PublishedThrough<Pricing> for Priced {}

struct Price;

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

The responder answers on the inbox the request named.

Macros

examples/requester.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // The broker's request policy: its live publisher correlates replies.
        b.include(price).out(Pricing, Request).build();
        b.include(quote_price).out_reply(Publish).transform(ToInbox);
    })
}

Manual

examples/manual/requester.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // The broker's request policy: its live publisher correlates replies.
        b.include(subscriber("orders", Price).build())
            .out(Pricing, Request)
            .build();
        b.include(
            subscriber("quotes.ask", QuoteAsk)
                .reply()
                .to("quotes.answers")
                .build(),
        )
        .out_reply(Publish)
        .transform(ToInbox);
    })
}