Ответы и публикация в другие темы

accept отвечает на каждый заказ запросом оплаты и уведомляет покупателя через слот Out. Роутер задаёт, как публикуются оба сообщения.

Подписчик публикует сообщения двумя способами. Ответ - это значение, которое возвращает тело обработчика. Рантайм кодирует его и публикует после успешного завершения тела. Слот Out - это издатель, которого тело получает параметром и вызывает само, столько раз, сколько нужно. Адрес сообщения - свойство его типа: #[outgoing(name = ..)] задаёт тему, поэтому ни тело, ни точка монтирования её не повторяют. Слот перечисляет типы, которые через него публикуются, поэтому публикация любого другого типа не компилируется. Точка монтирования связывает каждую позицию с политикой публикации, здесь это обычная политика брокера 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 публикует ответ и подтверждает сообщение, 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>. Тип 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(..) задаёт описание, которое макросу даёт doc-комментарий. include_router монтирует роутер на брокере. Приложение, которое это делает, показывает глава 5.