Headers

content-type, correlation-id, reply-to and message-id ride the AMQP properties section; every other header rides application-properties.

The crate adds no envelope of its own, so a peer that does not use RustStream reads a plain AMQP message. On publish each well-known header fills the property of the same name. The partition key fills group-id. Any other header becomes an application property: a UTF-8 value as a string, any other value as binary.

On delivery a handler reads both sections as one header map, through ctx.headers(). correlation_id() and reply_to() read the two properties request-reply uses. A message id or a correlation id the sender wrote as a UUID, a number or binary reads as text. An application property the sender wrote as a number, a boolean or a timestamp reads as text too; a timestamp reads as milliseconds.

A message type declares its headers with #[outgoing(headers = ..)], and a field renamed to correlation-id lands in the property. A HeaderMap with the same entries works the same. The in-process mode frames a publish through the same mapping, so a test reads the headers a live delivery carries.

Example

Macros

examples/amqp_publishing_headers.rs

/// The headers a shipment carries: `correlation-id` rides the `properties` section, `tenant` rides
/// `application-properties`.
#[derive(Debug, PartialEq, Deserialize, Serialize)]
struct Trace {
    #[serde(rename = "correlation-id")]
    correlation_id: String,
    tenant: String,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "shipments", headers = Trace)]
struct Shipment {
    order_id: u64,
}

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

#[subscriber(AmqpAddress::queue("orders"))]
async fn ship(
    order: &Order,
    ctx: &mut Context<'_>,
    Out(shipments): Out<impl Publisher, Shipments>,
) -> HandlerOutcome {
    // Both sections arrive as one header map: the property and the application property alike.
    let headers = ctx.headers();
    let trace = Trace {
        correlation_id: headers.correlation_id().unwrap_or_default().to_owned(),
        tenant: headers.get_str("tenant").unwrap_or_default().to_owned(),
    };
    tracing::info!(order.id, tenant = %trace.tenant, "shipping");
    let shipment = Shipment { order_id: order.id };
    if shipments
        .message(&shipment)
        .with_headers(&trace)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/amqp_publishing_headers.rs

/// The headers a shipment carries: `correlation-id` rides the `properties` section, `tenant` rides
/// `application-properties`.
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Trace {
    #[serde(rename = "correlation-id")]
    correlation_id: String,
    tenant: String,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "shipments", headers = Trace)]
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> {
        // Both sections arrive as one header map: the property and the application property alike.
        let headers = ctx.headers();
        let trace = Trace {
            correlation_id: headers.correlation_id().unwrap_or_default().to_owned(),
            tenant: headers.get_str("tenant").unwrap_or_default().to_owned(),
        };
        tracing::info!(order.id, tenant = %trace.tenant, "shipping");
        let shipment = Shipment { order_id: order.id };
        outs.get(Shipments)
            .message(&shipment)
            .with_headers(&trace)
            .publish()
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}