Headers

A publish's headers travel in the AMQP frame: content-type, correlation-id, reply-to and message-id ride the native properties of those names, and every other header lands in the header table.

A handler hands the headers to the publish builder with .with_headers(..), as a HeaderMap or as the typed contract its message declares. The four identity headers become properties, so a consumer written against another client finds them where the protocol puts them. Every other header is a byte string in the header table, so a binary value survives the round trip. A delivery reports both kinds back as headers, and the handler reads them from its context. A delivery also reports two properties under headers of their own: the priority under PRIORITY_HEADER (amqp-priority) and the TTL under EXPIRATION_HEADER (amqp-expiration). A header never sets a delivery property: priority, expiration and delivery mode are properties of the policy and the publish builder. An AMQP property is a short string of at most 255 bytes, and a longer identity header fails the publish with AmqpError::InvalidOptions before anything is routed.

Example

The invoice carries the order's reference as its message-id property and the tenant in the header table. The billing side reads both from its context.

Macros
Manual

examples/lapin_publishing_headers.rs
/// The invoice carries the order's reference as its `message-id` property and the tenant in the
/// header table.
#[subscriber(RabbitQueue::new("orders"))]
async fn issue_invoice(order: &Order, Out(invoices): Out<impl Publisher>) -> HandlerOutcome {
    let mut headers = HeaderMap::new();
    headers.insert("message-id", order.reference.clone());
    headers.insert("tenant", "acme");
    let invoice = Invoice { order_id: order.id };
    if let Err(err) = invoices
        .message(&invoice)
        .with_headers(headers)
        .publish()
        .await
    {
        // A reference longer than a short string is refused, and a retry will not shorten it.
        tracing::error!(order.id, %err, "invoice refused");
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

/// The billing side reads both where it reads every header.
#[subscriber(RabbitQueue::new("invoices"))]
async fn bill(invoice: &Invoice, ctx: &mut Context<'_>) -> HandlerOutcome {
    let reference = ctx.headers().get_str("message-id").unwrap_or_default();
    let tenant = ctx.headers().get_str("tenant").unwrap_or_default();
    tracing::info!(invoice.order_id, reference, tenant, "invoice billed");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(issue_invoice)
                .out(DefaultSlot, Publish::default())
                .build();
            b.include(bill);
        },
    )
}

examples/manual/lapin_publishing_headers.rs
/// The invoice carries the order's reference as its `message-id` property and the tenant in the
/// header table.
struct IssueInvoice;

impl<Egress> Handle<Order, (), Outs<(Egress,)>> for IssueInvoice
where
    Egress: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let mut headers = HeaderMap::new();
        headers.insert("message-id", order.reference.clone());
        headers.insert("tenant", "acme");
        let invoice = Invoice { order_id: order.id };
        let invoices = outs.get(DefaultSlot);
        if let Err(err) = invoices
            .message(&invoice)
            .with_headers(headers)
            .publish()
            .await
        {
            // A reference longer than a short string is refused, and a retry will not shorten it.
            tracing::error!(order.id, %err, "invoice refused");
            return Err(HandlerOutcome::drop());
        }
        Ok(())
    }
}

/// The billing side reads both where it reads every header.
struct Bill;

impl Handle<Invoice> for Bill {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let reference = ctx.headers().get_str("message-id").unwrap_or_default();
        let tenant = ctx.headers().get_str("tenant").unwrap_or_default();
        tracing::info!(invoice.order_id, reference, tenant, "invoice billed");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(subscriber(RabbitQueue::new("orders"), IssueInvoice).build())
                .out(DefaultSlot, Publish::default())
                .build();
            b.include(subscriber(RabbitQueue::new("invoices"), Bill).build());
        },
    )
}

Example​

Example