Replies and publishing to other topics

accept answers each order with a payment request and tells the customer through an Out slot; a router says how both leave.

A subscriber publishes in two ways. A reply is the value the body returns: the runtime encodes it and publishes it after the body succeeds. An Out slot is a publisher the body receives as a parameter and calls itself, as many times as it needs. Where a message goes is a property of its type: #[outgoing(name = ..)] fixes the topic, so neither the body nor the mount site repeats it. A slot lists the types that may leave through it, so publishing anything else does not compile. The mount site binds each position to a publish policy, here the broker's plain Publish.

The messages

Payments receives a PaymentRequest; Paid is what it answers in the next chapter.

Macros

payments.rs

/// What the orders side asks payments to charge.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema, Outgoing)]
#[outgoing(name = "payments.requested")]
pub(crate) struct PaymentRequest {
    pub(crate) order: u64,
    pub(crate) customer: String,
    pub(crate) lines: Vec<Line>,
    pub(crate) amount_cents: u64,
    pub(crate) request_id: String,
}

/// A charged order, on its way to shipping.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema, Outgoing)]
#[outgoing(name = "payments.settled")]
pub(crate) struct Paid {
    pub(crate) order: u64,
    pub(crate) customer: String,
    pub(crate) lines: Vec<Line>,
}

Manual

payments.rs

/// What the orders side asks payments to charge.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema)]
pub(crate) struct PaymentRequest {
    pub(crate) order: u64,
    pub(crate) customer: String,
    pub(crate) lines: Vec<Line>,
    pub(crate) amount_cents: u64,
    pub(crate) request_id: String,
}

// What `#[derive(Outgoing)]` with `#[outgoing(name = "payments.requested")]` writes.
impl OutgoingDestination for PaymentRequest {
    type Form = FixedName;

    const DESTINATION: &'static str = "payments.requested";
}

impl MessageHeaders for PaymentRequest {
    type Contract = NoHeaders;
}

impl MessageInfo for PaymentRequest {
    const NAME: &'static str = "PaymentRequest";
    const DESCRIPTION: Option<&'static str> = Some("What the orders side asks payments to charge.");
}

/// A charged order, on its way to shipping.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema)]
pub(crate) struct Paid {
    pub(crate) order: u64,
    pub(crate) customer: String,
    pub(crate) lines: Vec<Line>,
}

impl OutgoingDestination for Paid {
    type Form = FixedName;

    const DESTINATION: &'static str = "payments.settled";
}

impl MessageHeaders for Paid {
    type Contract = NoHeaders;
}

impl MessageInfo for Paid {
    const NAME: &'static str = "Paid";
    const DESCRIPTION: Option<&'static str> = Some("A charged order, on its way to shipping.");
}

A Notice is what the shop tells a customer, and the OutSlot marker Notices is the slot it leaves through. The manual form writes what #[derive(OutSlot)] generates: the slot's name, the types it publishes, and their entries in the AsyncAPI document.

Macros

notifications.rs

/// What the shop tells a customer.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema, Outgoing)]
#[outgoing(name = "notices")]
pub(crate) struct Notice {
    pub(crate) customer: String,
    pub(crate) text: String,
}

/// The slot a subscriber tells customers through, and what may leave through it.
#[derive(OutSlot)]
#[publishes(Notice)]
pub(crate) struct Notices;

Manual

notifications.rs

/// What the shop tells a customer.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema)]
pub(crate) struct Notice {
    pub(crate) customer: String,
    pub(crate) text: String,
}

// What `#[derive(Outgoing)]` with `#[outgoing(name = "notices")]` writes.
impl OutgoingDestination for Notice {
    type Form = FixedName;

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

impl MessageHeaders for Notice {
    type Contract = NoHeaders;
}

impl MessageInfo for Notice {
    const NAME: &'static str = "Notice";
    const DESCRIPTION: Option<&'static str> = Some("What the shop tells a customer.");
}

/// The slot a subscriber tells customers through, and what may leave through it.
pub(crate) struct Notices;

// What `#[derive(OutSlot)]` with `#[publishes(Notice)]` writes.
impl OutSlot for Notices {
    const NAME: &'static str = "Notices";
    // `Notice` fixes its destination, so no transform on this slot may name one.
    type Destination = Reads;

    fn outgoing() -> Vec<OutgoingMessageMetadata> {
        <Notice as OutMessages<Self>>::outgoing()
    }
}

impl PublishedThrough<Notices> for Notice {}

impl<Slot: OutSlot> OutMessages<Slot> for Notice {
    fn outgoing() -> Vec<OutgoingMessageMetadata> {
        vec![OutgoingMessageMetadata::new(
            Self::DESTINATION,
            type_name::<Self>(),
        )]
    }
}

The subscriber

accept drops an order with no lines, since there is nothing to charge, and an order past the shop's bounds, since it has no total. It tells the customer the order arrived, through the slot, and asks for a retry if that publish fails. Then it returns the payment request, which the runtime publishes as the reply. Ok publishes the reply and acks; Err publishes nothing and settles by the outcome.

Macros

orders.rs

/// Accepts an order: asks payments to charge it, and tells the customer it arrived.
#[subscriber("orders.placed", reply)]
async fn accept(
    order: &Order,
    Headers(meta): Headers<RequestMeta>,
    Out(notices): Out<impl Publisher, Notices>,
) -> Result<PaymentRequest, HandlerOutcome> {
    if order.lines.is_empty() {
        tracing::warn!(order = order.id, "an order with no lines, dropped");
        return Err(HandlerOutcome::drop());
    }
    let Some(amount_cents) = total_cents(&order.lines) else {
        tracing::warn!(order = order.id, "an order past the shop's bounds, dropped");
        return Err(HandlerOutcome::drop());
    };
    let notice = Notice {
        customer: order.customer.clone(),
        text: format!("We have your order {}.", order.id),
    };
    if notices.message(&notice).publish().await.is_err() {
        return Err(HandlerOutcome::retry());
    }
    Ok(PaymentRequest {
        order: order.id,
        customer: order.customer.clone(),
        lines: order.lines.clone(),
        amount_cents,
        request_id: meta.request_id,
    })
}

Manual

orders.rs

/// Accepts an order: asks payments to charge it, and tells the customer it arrived.
struct Accept;

impl<Entry> Handle<Message<RequestMeta, Order>, PaymentRequest, Outs<(Entry,)>, (), Shop> for Accept
where
    Entry: OutEntry<Notices, Wire: Publisher>,
{
    async fn handle(
        &self,
        placed: &Message<RequestMeta, Order>,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_, (), Shop>,
    ) -> Result<PaymentRequest, HandlerOutcome> {
        let (meta, order) = (&placed.headers, &placed.body);
        if order.lines.is_empty() {
            tracing::warn!(order = order.id, "an order with no lines, dropped");
            return Err(HandlerOutcome::drop());
        }
        let Some(amount_cents) = total_cents(&order.lines) else {
            tracing::warn!(order = order.id, "an order past the shop's bounds, dropped");
            return Err(HandlerOutcome::drop());
        };
        let notice = Notice {
            customer: order.customer.clone(),
            text: format!("We have your order {}.", order.id),
        };
        if outs.get(Notices).message(&notice).publish().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(PaymentRequest {
            order: order.id,
            customer: order.customer.clone(),
            lines: order.lines.clone(),
            amount_cents,
            request_id: meta.request_id.clone(),
        })
    }
}

In the manual form the slot is an entry of Outs, and outs.get(Notices) hands out its publisher.

The router

A Router collects one module's subscribers with the settings of their positions. out_reply binds the reply to a policy, and .codec(INTERNAL) encodes it as CBOR. out binds the slot; it keeps the router's codec, which is JSON here. The router names the broker and the app's state type in RouterDef<MemoryBroker, Shop>; chapter 7 introduces Shop.

Macros

orders.rs

/// The orders side's subscribers, with where their messages go.
pub(crate) fn routes() -> impl RouterDef<MemoryBroker, Shop> {
    Router::new()
        .include(audit)
        .include(accept)
        // The payment request travels as CBOR, the codec of the shop's own topics.
        .out_reply(Publish)
        .codec(INTERNAL)
        .out(Notices, Publish)
        .build()
}

Manual

orders.rs

/// The orders side's subscribers, with where their messages go.
pub(crate) fn routes() -> impl RouterDef<MemoryBroker, Shop> {
    Router::new()
        .include(
            subscriber(Order::DESTINATION, Audit)
                .describe(
                    "Writes every placed order to the audit log, with the request that placed it.",
                )
                .build(),
        )
        .include(
            subscriber(Order::DESTINATION, Accept)
                .describe(
                    "Accepts an order: asks payments to charge it, and tells the customer it \
                     arrived.",
                )
                .reply()
                .build(),
        )
        // The payment request travels as CBOR, the codec of the shop's own topics.
        .out_reply(Publish)
        .codec(INTERNAL)
        .out(Notices, Publish)
        .build()
}

The manual form binds each subscriber to its topic with subscriber(..), and .describe(..) gives it the description the doc comment gives the macro. include_router mounts the router on the broker; chapter 5 shows the app that does it.