PARTITION_KEY_HEADER

PARTITION_KEY_HEADER is the portable spelling of the FIFO message group: sent as MessageGroupId, read back on delivery, ignored by standard queues.

The header's name is partition-key, the one the core's in-memory broker uses, so a service that sets it runs unchanged on another broker. It never becomes a message attribute. A publish to a .fifo queue or topic sends it as the message group, and a delivery from a FIFO queue carries the group back in it. A standard queue has no message groups, so the header is dropped there, and a delivery from a standard queue carries none.

The group of a FIFO publish resolves from the most specific setting: the call's own group_id step, then this header, then the group the publish policy fixed. A FIFO publish that names none goes under the group default. SQS hands out one group's messages in order, and holds the group back while one of its messages is in flight, as the FIFO page describes.

Example

The router forwards each order to the packing line in the customer's group, and the packer reads the group back.

Macros
Manual

examples/sqs_delivery_partition_key.rs
/// Forwards each order to the packing line in the customer's message group, so one customer's
/// orders are packed in the order they were placed.
#[subscriber(SqsQueue::new("orders"))]
async fn route(order: &Order, Out(packing): Out<impl Publisher, Packing>) -> HandlerOutcome {
    let mut headers = HeaderMap::new();
    headers.insert(PARTITION_KEY_HEADER, order.customer.clone());
    match packing
        .message(order)
        .to("packing.fifo")
        .with_headers(headers)
        .publish()
        .await
    {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

/// Reads the group back from the delivery.
#[subscriber(SqsQueue::new("packing.fifo"))]
async fn pack(order: &Order, cx: &mut Context<'_>) -> HandlerOutcome {
    let group = cx
        .headers()
        .get(PARTITION_KEY_HEADER)
        .map(String::from_utf8_lossy);
    tracing::info!(order.id, ?group, "order packed");
    HandlerOutcome::ack()
}

examples/manual/sqs_delivery_partition_key.rs
/// Forwards each order to the packing line in the customer's message group, so one customer's
/// orders are packed in the order they were placed.
struct Route;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Route
where
    Entry: OutEntry<Packing, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let mut headers = HeaderMap::new();
        headers.insert(PARTITION_KEY_HEADER, order.customer.clone());
        let sent = outs
            .get(Packing)
            .message(order)
            .to("packing.fifo")
            .with_headers(headers)
            .publish()
            .await;
        sent.map_err(|_| HandlerOutcome::retry())
    }
}

/// Reads the group back from the delivery.
struct Pack;

impl Handle<Order> for Pack {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let group = ctx
            .headers()
            .get(PARTITION_KEY_HEADER)
            .map(String::from_utf8_lossy);
        tracing::info!(order.id, ?group, "order packed");
        ready(Ok(()))
    }
}

Example​

Example