Headers

A serde struct is the header contract a handler reads and a publish writes, checked in both directions.

A contract is a flat struct: each field names a header, and its value travels as a string. Fields are numbers, booleans, strings, bytes, unit-only enums, or Options of those. An Option field reads as None when its header is absent; any other missing field fails the contract. Header names are matched without regard to case; #[serde(rename = "..")] gives a wire name that is not a Rust identifier. Headers<T> parses the contract before the body runs. A contract that does not parse is a decode failure, settled by the subscriber's on_failure(decode = ..) policy. In a manual body Headers::<T>::from_context(ctx) parses it, and a failure there drops the delivery. The raw HeaderMap of a delivery stays on its context. What the handler publishes carries a contract of its own, declared on the outgoing type.

Example

Macros

examples/typed_headers.rs

/// What travels in the headers of a chunk: each field is one header, its value a string.
#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct ChunkMeta {
    task_id: u64,
    chunk_no: u32,
    chunks_total: u32,
}

/// What travels in the headers of a finished chunk.
#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct DoneMeta {
    task_id: u64,
    chunk_no: u32,
}

Manual

examples/manual/typed_headers.rs

/// What travels in the headers of a chunk: each field is one header, its value a string.
#[derive(Debug, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
struct ChunkMeta {
    task_id: u64,
    chunk_no: u32,
    chunks_total: u32,
}

/// What travels in the headers of a finished chunk.
#[derive(Debug, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
struct DoneMeta {
    task_id: u64,
    chunk_no: u32,
}

Macros

examples/typed_headers.rs

/// The contract is parsed before the body runs; what the body publishes carries its own.
#[subscriber("chunks")]
async fn convert(
    chunk: &Chunk,
    Headers(meta): Headers<ChunkMeta>,
    Out(events): Out<impl Publisher, Events>,
) -> HandlerOutcome {
    tracing::info!(meta.task_id, meta.chunk_no, chunk.size, "converting");
    let done = ChunkDone {
        output_key: format!("chunks/{}/{}.part", meta.task_id, meta.chunk_no),
    };
    let done_meta = DoneMeta {
        task_id: meta.task_id,
        chunk_no: meta.chunk_no,
    };
    match events
        .message(&done)
        .with_headers(&done_meta)
        .publish()
        .await
    {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

Manual

examples/manual/typed_headers.rs

/// The contract is parsed before the work starts; what the body publishes carries its own.
struct Convert;

impl<Entry> Handle<Chunk, (), Outs<(Entry,)>> for Convert
where
    Entry: OutEntry<Events, Wire: Publisher>,
{
    async fn handle(
        &self,
        chunk: &Chunk,
        outs: &Outs<(Entry,)>,
        ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let Headers(meta) = Headers::<ChunkMeta>::from_context(ctx).await?;
        tracing::info!(meta.task_id, meta.chunk_no, chunk.size, "converting");
        let done = ChunkDone {
            output_key: format!("chunks/{}/{}.part", meta.task_id, meta.chunk_no),
        };
        let done_meta = DoneMeta {
            task_id: meta.task_id,
            chunk_no: meta.chunk_no,
        };
        let sent = outs
            .get(Events)
            .message(&done)
            .with_headers(&done_meta)
            .publish()
            .await;
        sent.map_err(|_| HandlerOutcome::retry())
    }
}