Replies

A handler answers a message by returning a value; the reply type or the mount site says where the answer goes.

Every reply type derives Outgoing. A type with #[outgoing(name = "..")] fixes its destination, and the handler carries the bare clause #[subscriber("src", reply)]. A type that names nothing takes the destination from the clause, reply("dest"). A destination declared on the type wins over a name at the mount site, and startup logs a warning for such a registration. The forms of a destination are on the Destinations page. Result<Reply, HandlerOutcome> keeps the settlement in the handler's hands: Ok publishes the reply and acks, Err publishes nothing and settles by the outcome. A reply whose publish fails nacks the delivery with requeue, so a replying handler has to be safe to run twice. A batch handler returns one Vec of replies per batch.

With nothing else said, the reply leaves through the broker's default policy under the default codec. .out_reply(policy) names the publish policy, and the steps after it complete the reply's wiring. .codec(..) names the codec of the reply, while the request still decodes with the scope's codec. .transform(..) adds a publish transform, and .transactional() puts every reply of a batch in one broker transaction. A broker that only reads the payload is lent the dispatch loop's own encode buffer, so a reply allocates nothing per delivery.

Example

The reply type fixes its destination.

Macros
Manual

examples/replies.rs
/// The reply type fixes its destination, so the clause is the bare `reply`.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

/// `Ok` publishes the receipt and acks; `Err` publishes nothing and settles by the outcome.
#[subscriber("orders", reply)]
async fn issue_receipt(order: &Order) -> Result<Receipt, HandlerOutcome> {
    if order.id == 0 {
        return Err(HandlerOutcome::drop());
    }
    Ok(Receipt { order_id: order.id })
}

examples/manual/replies.rs
/// The reply type fixes its destination, so the chain names none.
#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Receipt {
    order_id: u64,
}

// `#[outgoing(name = "receipts")]` by hand.
impl OutgoingDestination for Receipt {
    type Form = FixedName;

    const DESTINATION: &'static str = "receipts";
}

impl MessageHeaders for Receipt {
    type Contract = NoHeaders;
}

/// `Ok` publishes the receipt and acks; `Err` publishes nothing and settles by the outcome.
struct IssueReceipt;

impl Handle<Order, Receipt> for IssueReceipt {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        if order.id == 0 {
            return ready(Err(HandlerOutcome::drop()));
        }
        ready(Ok(Receipt { order_id: order.id }))
    }
}

The clause names the destination of a type that declares none.

Macros
Manual

examples/replies.rs
/// The reply type declares no destination, so the clause names one.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct Quote {
    order_id: u64,
    price: u64,
}

#[subscriber("quote-requests", reply("quotes"))]
async fn quote(order: &Order) -> Quote {
    Quote {
        order_id: order.id,
        price: order.id * 10,
    }
}

examples/manual/replies.rs
/// The reply type declares no destination, so the chain names one.
#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Quote {
    order_id: u64,
    price: u64,
}

impl OutgoingDestination for Quote {
    type Form = CallerName;
}

impl MessageHeaders for Quote {
    type Contract = NoHeaders;
}

struct QuoteOrder;

impl Handle<Order, Quote> for QuoteOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Quote, HandlerOutcome>> {
        ready(Ok(Quote {
            order_id: order.id,
            price: order.id * 10,
        }))
    }
}

Macros
Manual

examples/replies.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // The broker's default policy, under the default codec.
        b.include(issue_receipt);
        // The policy the reply leaves through, named at the mount site.
        b.include(quote).out_reply(Publish);
    })
}

examples/manual/replies.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // The broker's default policy, under the default codec.
        b.include(subscriber("orders", IssueReceipt).reply().build());
        // The destination the type leaves open, and the policy the reply leaves through.
        b.include(
            subscriber("quote-requests", QuoteOrder)
                .reply()
                .to("quotes")
                .build(),
        )
        .out_reply(Publish);
    })
}

Example​

Example