回复与发布到其他主题

accept 用一个支付请求回复每个订单，并通过 Out 槽位通知顾客。路由器决定两者如何发出。

订阅者有两种发布方式。 回复是主体返回的值：主体成功之后，运行时对它编码并发布。 Out 槽位是一个发布者。主体把它作为参数接收，并自己调用，需要多少次就调用多少次。 消息发往哪里是消息类型的属性：#[outgoing(name = ..)] 固定主题，主体和挂载点都不必重复它。 槽位列出可以经由它发出的类型，发布其他任何类型都无法编译。 挂载点把每个发布位置绑定到一个发布策略，这里是 Broker 的普通 Publish。

消息

支付模块接收 PaymentRequest，Paid 是它在下一章给出的回复。

宏

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>,
}

手写

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.");
}

Notice 是商店发给顾客的通知，OutSlot 标记 Notices 是它发出所经由的槽位。 手写形式写出 #[derive(OutSlot)] 生成的内容：槽位的名字、它发布的类型，以及这些类型在 AsyncAPI 文档中的条目。

宏

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;

手写

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>(),
        )]
    }
}

订阅者

没有订单行的订单，accept 直接丢弃，因为没有可收款的内容；超出商店上限的订单也会丢弃，因为它没有总额。 它通过槽位告诉顾客订单已收到，如果这次发布失败，就请求重试。 然后它返回支付请求，运行时把它作为回复发布。 Ok 发布回复并 ack，Err 不发布任何内容，按结果结算。

宏

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,
    })
}

手写

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(),
        })
    }
}

在手写形式中，槽位是 Outs 中的一个条目，outs.get(Notices) 交出它的发布者。

路由器

Router 汇集一个模块的订阅者，以及它们各个发布位置的设置。 out_reply 把回复绑定到一个策略，.codec(INTERNAL) 用 CBOR 对它编码。 out 绑定槽位。槽位沿用路由器的编解码器，这里是 JSON。 路由器在 RouterDef<MemoryBroker, Shop> 中指明 Broker 和应用的状态类型，Shop 在第 7 章介绍。

宏

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()
}

手写

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()
}

手写形式用 subscriber(..) 把每个订阅者绑定到它的主题，.describe(..) 给它描述，相当于宏从文档注释得到的描述。 include_router 把路由器挂载到 Broker 上，第 5 章展示完成这一步的应用。