Skip to main content

Messages: payloads, headers, codecs

An order grows into a payload with lines, a header contract every publish must carry, and a codec chosen per topic.

The payload is any serde type; the codec turns it into bytes and back. The headers travel beside the payload, and a header contract is a serde struct with one field per header. A message type that names a contract cannot be published without it: the publish does not compile. A subscriber reads the contract with a Headers<T> parameter, parsed before the body runs. A delivery whose headers do not parse never reaches the body and is dropped, like a payload that does not decode. The codec belongs to the mount site, not to the type, so one type can travel as JSON on one topic and as CBOR on another.

The order​

From here on, the shop's code lives in modules. orders.rs holds the order: its lines, the total, and the contract it travels with. The bounds cap what one order may ask for, and the total adds up with checked arithmetic, so an order past them has no total.

orders.rs
/// How many lines an order may hold.
pub(crate) const MAX_LINES: usize = 100;
/// How many items one line may order.
pub(crate) const MAX_QUANTITY: u32 = 1_000;
/// The highest price of one item, in cents.
pub(crate) const MAX_PRICE_CENTS: u64 = 1_000_000;

/// One line of an order: what the customer buys, how many, and at what price.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema)]
pub(crate) struct Line {
    pub(crate) sku: String,
    pub(crate) quantity: u32,
    pub(crate) price_cents: u64,
}

/// An order placed in the storefront. It travels with the headers of the request that placed it.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema, Outgoing)]
#[outgoing(name = "orders.placed", headers = RequestMeta)]
pub(crate) struct Order {
    pub(crate) id: u64,
    pub(crate) customer: String,
    pub(crate) lines: Vec<Line>,
}

/// Whether `lines` stay within the shop's bounds.
pub(crate) fn within_bounds(lines: &[Line]) -> bool {
    lines.len() <= MAX_LINES
        && lines
            .iter()
            .all(|line| line.quantity <= MAX_QUANTITY && line.price_cents <= MAX_PRICE_CENTS)
}

/// What `lines` cost together, or `None` past the shop's bounds.
pub(crate) fn total_cents(lines: &[Line]) -> Option<u64> {
    if !within_bounds(lines) {
        return None;
    }
    lines.iter().try_fold(0_u64, |total, line| {
        total.checked_add(u64::from(line.quantity).checked_mul(line.price_cents)?)
    })
}

headers = RequestMeta makes the contract part of the type. The manual form says the same with type Contract = WithHeaders<RequestMeta>, and gives the message its name in the AsyncAPI document with a MessageInfo impl.

The header contract​

Each field of the contract is one header. request_id names the storefront request that placed the order, so every step can log it.

orders.rs
/// The headers of a placed order: each field is one header.
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema)]
pub(crate) struct RequestMeta {
    pub(crate) request_id: String,
}

Reading the headers​

audit is a second subscriber on orders.placed: it writes every order to the audit log with the request that placed it. Each subscription gets every message of its topic, so audit and the accept of the next chapter both see each order. The manual form takes the pair Message<RequestMeta, Order> as its input: both parts decode before the body runs, and the AsyncAPI document shows the header schema.

orders.rs
/// Writes every placed order to the audit log, with the request that placed it.
#[subscriber("orders.placed")]
async fn audit(order: &Order, Headers(meta): Headers<RequestMeta>) -> HandlerOutcome {
    tracing::info!(
        request = %meta.request_id,
        order = order.id,
        total_cents = total_cents(&order.lines),
        "order placed",
    );
    HandlerOutcome::ack()
}

The codec of the shop's own topics​

JSON is the default codec, and the shop keeps it for orders.placed and notices, which other teams read. The topics between the shop's own steps travel as CBOR: smaller, and read by no one else. One constant names that codec, so the routers of the next chapters cannot disagree about it.

payments.rs
/// The codec of the topics between the shop's own steps: CBOR is smaller than JSON, and no
/// other team reads them.
pub(crate) const INTERNAL: CborCodec = CborCodec;

A router decodes with the codec with_codec names, and each publish position names its own with .codec(..). The next chapter puts both to use. CBOR needs the crate's cbor feature.