InboxHeaders

In the headers layout a headers struct describes the queue table and its header fields, and the message struct that flattens it is what a handler takes.

Two structs describe one queue. The headers struct derives InboxHeaders and describes the table: #[inbox(table, schema, advisory_lock, clock, isolation, mode)] and the fields that play a role, so it also selects the form. Its other fields are the delivery's header values. The message struct derives Inbox and holds the headers struct in a field marked #[field(headers)] and #[sqlx(flatten)], beside data of its own. It takes #[inbox(custom(..))] alone, and a role or a table attribute on it does not compile, with an error that names the headers struct as the place for it.

The headers layout is row mode: a handler takes the message struct itself, or a slice of them in a batch. Several message structs may flatten one headers struct, each read by its own subscriptions. The claim selects the ids of the rows it takes, and the default fetch reads their rows from the queue table, the headers struct's columns and the message's own fields. A subscription prepares that fetch when it opens, so a message field whose column the table lacks stops startup with SqlxBrokerError::Schema. A message whose data lives in other tables reads it with a fetch of its own.

The headers struct gets the generated insert, and the message struct gets none. By hand, one description on the message holds the whole table: .data(..) names the header columns, .fetching(..) the message's own, and .header_fields() builds the header map through HeaderFields and put_header.

Example

A packing job: OrderHeaders describes order_jobs and its headers tenant, trace and order_id, and OrderJob adds the job's note.

Macros
Manual

examples/sqlx_headers_layout.rs
/// The queue table: the columns that run the queue, and the headers `tenant`, `trace` and
/// `order_id`.
#[derive(Debug, Clone, InboxHeaders, sqlx::FromRow)]
#[inbox(table = "order_jobs")]
struct OrderHeaders {
    #[field(id, generated)]
    job_id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    tenant: String,
    trace: Option<String>,
    order_id: i64,
}

/// A job and its note, read from the queue table by the default fetch.
#[derive(Debug, Clone, Inbox, sqlx::FromRow)]
struct OrderJob {
    #[field(headers)]
    #[sqlx(flatten)]
    headers: OrderHeaders,
    note: Option<String>,
}

// The header map is built from `OrderHeaders` on its first read, here by `Headers<Tenant>`.
#[subscriber(InboxQueue::<OrderJob>::new("packing"))]
async fn pack(job: &OrderJob, Headers(tenant): Headers<Tenant>) -> HandlerOutcome {
    tracing::info!(
        tenant = %tenant.tenant,
        order = job.headers.order_id,
        note = ?job.note,
        "packing"
    );
    HandlerOutcome::ack()
}

examples/manual/sqlx_headers_layout.rs
use ruststream::HeaderMap;
use ruststream::runtime::SoloCarried;
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{self, Attempt, Lease};
use ruststream_sqlx::{AttemptRow, HeaderFields, InboxSpec, InboxTable, put_header};

#[derive(Debug, Clone, sqlx::FromRow, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct OrderHeaders {
    job_id: i64,
    attempt: i16,
    tenant: String,
    trace: Option<String>,
    order_id: i64,
}

#[derive(Debug, Clone, sqlx::FromRow, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct OrderJob {
    #[sqlx(flatten)]
    headers: OrderHeaders,
    note: Option<String>,
}

impl InboxTable for OrderJob {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, Attempt, spec::HeaderFields)>;
    const TABLE: Self::Table = InboxSpec::new("order_jobs", Column::new("job_id").generated())
        .lease(Column::new("locked_until"))
        .attempt(Column::new("attempt").generated())
        .data(&[
            Column::new("tenant"),
            Column::new("trace"),
            Column::new("order_id"),
        ])
        .fetching(&[Column::new("note")])
        .header_fields();

    fn id(&self) -> &i64 {
        &self.headers.job_id
    }
}

impl Input for OrderJob {
    type Axis = SoloCarried<Self>;
}

impl AttemptRow for OrderJob {
    type Attempt = i16;

    fn attempt(&self) -> &i16 {
        &self.headers.attempt
    }
}

impl HeaderFields for OrderJob {
    const NAMES: &'static [&'static str] = &["tenant", "trace", "order_id"];

    fn header_map(&self) -> HeaderMap {
        let mut headers = HeaderMap::with_capacity(Self::NAMES.len());
        put_header(&mut headers, "tenant", &self.headers.tenant);
        put_header(&mut headers, "trace", &self.headers.trace);
        put_header(&mut headers, "order_id", &self.headers.order_id);
        headers
    }
}

struct Packer;

impl Handle<OrderJob> for Packer {
    fn handle(
        &self,
        job: &OrderJob,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        // The header map is built from the fields on its first read.
        let tenant = ctx.headers().get_str("tenant").unwrap_or_default();
        tracing::info!(tenant, order = job.headers.order_id, note = ?job.note, "packing");
        ready(Ok(()))
    }
}

Example​

Example