Headers

The header frame is UTF-8 name: value lines; a header the frame cannot carry is refused on publish, and a malformed frame surfaces as ZmqError::Wire.

Frame 1 holds one header per line, the name and the value separated by :, and lines separated by \n. A service reads and writes headers through the core's typed contract: a serde struct whose fields are the headers, parsed by Headers<T> in the handler and declared on an outgoing type with #[outgoing(headers = T)].

Nothing is guessed in either direction. A header value that is not UTF-8 has no form in a text frame, so its publish fails with ZmqError::Send, naming the header and the destination. So does a header that would read back as something else: a line break in its name or value, a : in its name, or whitespace at either end of either. A peer's header frame that is not UTF-8, or a line without :, fails the message as ZmqError::Wire on the subscription, and the message never reaches the handler. Blank lines are skipped, so a peer that ends the frame with a newline interoperates. A name and a value are trimmed of surrounding whitespace on receipt.

Peers in other languages composes the header frame from a Python peer.

Example

A Python peer writes the frame by hand:

socket.send_multipart([b"jobs", b"x-tenant: acme\ncontent-type: application/json", payload])

The worker reads the tenant a peer names in x-tenant and sends it on with the result.

Macros
Manual

examples/zmq_wire_headers.rs
/// The header line `x-tenant: acme`, as a contract both directions check.
#[derive(Debug, PartialEq, Deserialize, Serialize)]
struct Tenant {
    #[serde(rename = "x-tenant")]
    tenant: String,
}

/// A result cannot be published without its tenant.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "results", headers = Tenant)]
struct Done {
    id: u64,
}

#[derive(OutSlot)]
#[publishes(Done)]
struct Results;

#[subscriber("jobs")]
async fn work(
    job: &Job,
    Headers(tenant): Headers<Tenant>,
    Out(results): Out<impl Publisher, Results>,
) -> HandlerOutcome {
    tracing::info!(job.id, tenant = %tenant.tenant, "working");
    match results
        .message(&Done { id: job.id })
        .with_headers(&tenant)
        .publish()
        .await
    {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(subscription = "jobs", job.id, error = %err, "result not published");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/zmq_wire_headers.rs
/// The header line `x-tenant: acme`, as a contract both directions check.
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Tenant {
    #[serde(rename = "x-tenant")]
    tenant: String,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Done {
    id: u64,
}

// `#[outgoing(name = "results", headers = Tenant)]` by hand: a result cannot be published
// without its tenant.
impl OutgoingDestination for Done {
    type Form = FixedName;
    const DESTINATION: &'static str = "results";
}

impl MessageHeaders for Done {
    type Contract = WithHeaders<Tenant>;
}

/// The slot results leave through.
struct Results;

impl OutSlot for Results {
    const NAME: &'static str = "Results";
    // `Done` fixes its destination, so no transform on this slot may name one.
    type Destination = Reads;
}

impl PublishedThrough<Results> for Done {}

struct Work;

impl<Entry> Handle<Job, (), Outs<(Entry,)>> for Work
where
    Entry: OutEntry<Results, Wire: Publisher>,
{
    async fn handle(
        &self,
        job: &Job,
        outs: &Outs<(Entry,)>,
        ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let Headers(tenant) = Headers::<Tenant>::from_context(ctx).await?;
        tracing::info!(job.id, tenant = %tenant.tenant, "working");
        let done = Done { id: job.id };
        if let Err(err) = outs
            .get(Results)
            .message(&done)
            .with_headers(&tenant)
            .publish()
            .await
        {
            tracing::warn!(subscription = "jobs", job.id, error = %err, "result not published");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Example​

Example