Keys

The partition-key header, PARTITION_KEY_HEADER, rides the AMQP group-id property, and a broker that honours message groups keeps one group on one consumer.

The header is named partition-key, the name the in-memory broker uses, so a service keeps its headers when it changes brokers. On publish its value becomes the message's group-id and does not travel as an application property. On delivery group-id comes back as the partition-key header.

Message groups are the broker's feature. On ActiveMQ Artemis every message of one group goes to one consumer of the queue, so the consumer sees the group in the order it was sent. Competing consumers of an AmqpAddress::queue address then share the groups instead of the messages.

A message type declares the header with #[outgoing(headers = ..)] and a field renamed to partition-key; a HeaderMap with that entry works the same. In process the key travels as on a server, and the consumers of a queue take turns whatever the group. Pinning a group to one consumer is the server's work, so a running broker is where it shows.

Example

Macros
Manual

examples/amqp_publishing_keys.rs
/// The header a shipment carries: `partition-key`, which becomes the `group-id` property.
#[derive(Debug, PartialEq, Deserialize, Serialize)]
struct ShipmentKey {
    #[serde(rename = "partition-key")]
    customer: String,
}

/// Every shipment of one customer is one message group, so one packer takes all of them.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "shipments", headers = ShipmentKey)]
struct Shipment {
    order_id: u64,
}

#[derive(OutSlot)]
#[publishes(Shipment)]
struct Shipments;

#[subscriber(AmqpAddress::queue("orders"))]
async fn ship(order: &Order, Out(shipments): Out<impl Publisher, Shipments>) -> HandlerOutcome {
    let key = ShipmentKey {
        customer: order.customer.clone(),
    };
    tracing::info!(order.id, customer = %key.customer, "shipping");
    let shipment = Shipment { order_id: order.id };
    if shipments
        .message(&shipment)
        .with_headers(&key)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/amqp_publishing_keys.rs
/// The header a shipment carries: `partition-key`, which becomes the `group-id` property.
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct ShipmentKey {
    #[serde(rename = "partition-key")]
    customer: String,
}

/// Every shipment of one customer is one message group, so one packer takes all of them.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "shipments", headers = ShipmentKey)]
struct Shipment {
    order_id: u64,
}

#[derive(OutSlot)]
#[publishes(Shipment)]
struct Shipments;

struct Ship;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Ship
where
    Entry: OutEntry<Shipments, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let key = ShipmentKey {
            customer: order.customer.clone(),
        };
        tracing::info!(order.id, customer = %key.customer, "shipping");
        let shipment = Shipment { order_id: order.id };
        outs.get(Shipments)
            .message(&shipment)
            .with_headers(&key)
            .publish()
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

Example​

Example