Headers

A header travels as a Pulsar message property, one per header and no envelope; a value that is not UTF-8 fails the publish.

The property has the header's name and value, so a consumer in another language reads it as a plain Pulsar property. The payload stays the encoded message alone. A message type declares its headers in a typed contract, #[outgoing(headers = ..)], and the handler sets them with with_headers. A HeaderMap written by hand works the same. Pulsar properties are text, so a header value that is not valid UTF-8 fails the publish with PulsarError::Publish. On delivery the properties come back as the message's headers. One header has a meaning of its own: the partition-key header becomes the message key.

Publishing and replies sends pick lists with headers to warehouses.

Example

Macros

examples/pulsar_publishing_headers.rs

/// The header contract of an invoice: the tenant it belongs to travels as the `tenant` property.
#[derive(Debug, PartialEq, Deserialize, Serialize)]
struct Tenant {
    tenant: String,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "invoices", headers = Tenant)]
struct Invoice {
    order_id: u64,
}

#[derive(OutSlot)]
#[publishes(Invoice)]
struct Invoices;

/// Issues the invoice of each order under its tenant.
#[subscriber(PulsarSubscription::new("orders", "billing"))]
async fn issue(order: &Order, Out(invoices): Out<impl Publisher, Invoices>) -> HandlerOutcome {
    let tenant = Tenant {
        tenant: order.tenant.clone(),
    };
    let sent = invoices
        .message(&Invoice { order_id: order.id })
        .with_headers(&tenant)
        .publish()
        .await;
    match sent {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(order.id, %err, "the invoice was not issued");
            HandlerOutcome::retry()
        }
    }
}

Manual

examples/manual/pulsar_publishing_headers.rs

/// The header contract of an invoice: the tenant it belongs to travels as the `tenant` property.
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
struct Tenant {
    tenant: String,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[outgoing(name = "invoices", headers = Tenant)]
struct Invoice {
    order_id: u64,
}

#[derive(OutSlot)]
#[publishes(Invoice)]
struct Invoices;

/// Issues the invoice of each order under its tenant.
struct Issue;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Issue
where
    Entry: OutEntry<Invoices, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let tenant = Tenant {
            tenant: order.tenant.clone(),
        };
        outs.get(Invoices)
            .message(&Invoice { order_id: order.id })
            .with_headers(&tenant)
            .publish()
            .await
            .map_err(|err| {
                tracing::warn!(order.id, %err, "the invoice was not issued");
                HandlerOutcome::retry()
            })
    }
}