PARTITION_KEY_HEADER

The partition-key header is the portable spelling of the ordering key, read on publish and written on every keyed delivery.

A body that names no broker type sets the key through this header. On publish the header becomes the message's ordering key and does not travel as an attribute. The header names the same key as the in-memory broker's partition-key, so a service keeps its headers when it moves between the two. A key from the ordering_key step or a transform wins over the header, and the header wins over the policy's key. A value that is not UTF-8 fails the publish. On a delivery the header holds the key the message was published under, and it is absent on a message without one.

Example

The handler sets the header, and the shipment goes out under the order's key.

Macros
Manual

examples/pubsub_publishing_ordering_header.rs
/// The body names no broker type: the `partition-key` header orders the shipment.
#[subscriber("orders-workers")]
async fn ship(order: &Order, Out(out): Out<impl Publisher>) -> HandlerOutcome {
    let mut headers = HeaderMap::new();
    headers.insert(PARTITION_KEY_HEADER, format!("order-{}", order.id));
    let shipment = Shipment { order_id: order.id };
    let sent = out
        .message(&shipment)
        .with_headers(headers)
        .to("shipments")
        .publish()
        .await;
    match sent {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

examples/manual/pubsub_publishing_ordering_header.rs
/// The body names no broker type: the `partition-key` header orders the shipment.
struct Ship;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Ship
where
    Entry: OutEntry<DefaultSlot, 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, format!("order-{}", order.id));
        let shipment = Shipment { order_id: order.id };
        let sent = outs
            .get(DefaultSlot)
            .message(&shipment)
            .with_headers(headers)
            .to("shipments")
            .publish()
            .await;
        sent.map_err(|_| HandlerOutcome::retry())
    }
}

Example​

Example