Headers

Headers travel as Pub/Sub attributes, with no envelope of their own; an attribute that is not UTF-8 fails the publish.

Every header of an outgoing message becomes an attribute of the Pub/Sub message, name and value unchanged. Each attribute of a delivery becomes a header, so a handler reads it with ctx.headers(). Attributes are text, so a header value that is not UTF-8 fails the publish with PubSubError::Publish instead of being rewritten. Pub/Sub accepts at most 100 attributes per message, names of up to 256 bytes and values of up to 1024 bytes, and refuses a message past them.

Two headers have a meaning of their own. partition-key is the message's ordering key: it never travels as an attribute, and a delivery reports its ordering key under it. pubsub-delivery-attempt is the attempt number a delivery reports where the subscription counts attempts.

Example

The tenant arrives as an attribute of the order and leaves as an attribute of the shipment.

Macros

examples/pubsub_publishing_headers.rs

/// The tenant arrives as an attribute of the order and leaves as an attribute of the shipment.
#[subscriber("orders-workers")]
async fn ship(
    order: &Order,
    ctx: &mut Context<'_>,
    Out(out): Out<impl Publisher>,
) -> HandlerOutcome {
    let mut headers = HeaderMap::new();
    if let Some(tenant) = ctx.headers().get_str("tenant") {
        headers.insert("tenant", tenant.to_owned());
    }
    let shipment = Shipment { order_id: order.id };
    match out.message(&shipment).with_headers(headers).publish().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

Manual

examples/manual/pubsub_publishing_headers.rs

/// The tenant arrives as an attribute of the order and leaves as an attribute of 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();
        if let Some(tenant) = ctx.headers().get_str("tenant") {
            headers.insert("tenant", tenant.to_owned());
        }
        let shipment = Shipment { order_id: order.id };
        let sent = outs
            .get(DefaultSlot)
            .message(&shipment)
            .with_headers(headers)
            .publish()
            .await;
        sent.map_err(|_| HandlerOutcome::retry())
    }
}