Headers

Headers travel inside the payload: a message with headers is written as rs1: or rs2: plus base64, one without is written as it is.

A stream file entry and a line on a pipe carry a stream key and a payload, with no room for headers. So a message with headers is written as an envelope: a prefix, then base64 of a header block followed by the payload. The prefix rs1: marks a block of name: value lines. It is used whenever every header reads back unchanged from a line, which keeps the message readable by earlier releases of this crate. The prefix rs2: marks a block where each name and value is length-prefixed. It is used when a value is not UTF-8, holds a line break, or begins or ends with whitespace, so a value travels byte for byte. The envelope is text, so it fits on a line of a pipe. A message with no headers is written as it is, so a file recorded that way stays readable by any sea-streamer consumer, and a file another tool wrote is readable here. A payload that itself begins with rs1: or rs2: is enveloped, so no reader takes it for an envelope. The reader opens the envelope before the handler runs, so a handler reads headers with Headers<T> or ctx.headers() as on any broker. Every delivery also carries its position in SEQUENCE_HEADER.

Example

The service confirms each order under the tenant header it arrived with, and the receipt carries the same header.

Macros

examples/sea_headers.rs

/// The header contract every order and receipt carries.
#[derive(Debug, PartialEq, Deserialize, Serialize)]
struct Tenant {
    tenant: String,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders", headers = Tenant)]
struct Order {
    id: u64,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts", headers = Tenant)]
struct Receipt {
    order: u64,
}

#[derive(OutSlot)]
#[publishes(Receipt)]
struct Receipts;

/// Confirms an order under the tenant it arrived with.
#[subscriber(FileStream::new("orders"))]
async fn confirm(
    order: &Order,
    Headers(tenant): Headers<Tenant>,
    Out(receipts): Out<impl Publisher, Receipts>,
) -> HandlerOutcome {
    tracing::info!(order.id, tenant = %tenant.tenant, "confirmed");
    let receipt = Receipt { order: order.id };
    if receipts
        .message(&receipt)
        .with_headers(&tenant)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// The service on the file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker(file, |b| {
        b.include(confirm).out(Receipts, Publish).build();
    })
}

Manual

examples/manual/sea_headers.rs

/// The header contract every order and receipt carries.
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Tenant {
    tenant: String,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "orders", headers = Tenant)]
struct Order {
    id: u64,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "receipts", headers = Tenant)]
struct Receipt {
    order: u64,
}

#[derive(OutSlot)]
#[publishes(Receipt)]
struct Receipts;

/// Confirms an order under the tenant it arrived with.
struct Confirm;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Confirm
where
    Entry: OutEntry<Receipts, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let Headers(tenant) = Headers::<Tenant>::from_context(ctx).await?;
        tracing::info!(order.id, tenant = %tenant.tenant, "confirmed");
        let receipt = Receipt { order: order.id };
        if outs
            .get(Receipts)
            .message(&receipt)
            .with_headers(&tenant)
            .publish()
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

/// The service on the file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("checkout", "0.1.0")).with_broker(file, |b| {
        b.include(subscriber(FileStream::new("orders"), Confirm).build())
            .out(Receipts, Publish)
            .build();
    })
}