Headers

A headers column keeps a delivery's headers as a JSON object of strings; a publish with a header the column cannot hold is refused.

#[field(headers)] marks the column, of type Option<Json<BTreeMap<String, String>>> (feature json); by hand .headers(..) with HeaderRow, which lends the field. A claim moves the column's object into the delivery's header map, so middleware, ctx.headers() and the Headers<T> extractor read it as on any broker. In row mode the row a handler reads holds the column empty.

The service's Publish writes the column, with HeaderColumn::from_headers over the message's headers. A JSON string holds UTF-8 text only, so a header value that is not UTF-8 would come back changed. Such a publish fails with SqlxBrokerError::Header before the service's Publish runs, and a header never reaches the table changed. A table without a headers column holds no header at all, and refuses a publish that carries one. A column of the service's own type implements HeaderColumn, and a table whose headers are columns of their own takes the headers layout.

Example

The tenant of an export rides the headers column: the handler reads it with Headers<Tenant> and drops an export for a tenant the service does not serve.

Macros
Manual

examples/sqlx_headers.rs
use ruststream_sqlx::HeaderColumn;
use sqlx::types::Json;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "export_jobs")]
struct ExportJob {
    #[field(id, generated)]
    id: i64,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(headers)]
    meta: Option<Json<BTreeMap<String, String>>>,
    #[field(payload)]
    payload: Vec<u8>,
}

impl Publish<Sqlite> for ExportJob {
    async fn publish(
        conn: &mut SqliteConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        // The column keeps the message's headers as the delivery reads them back.
        let headers = Json::<BTreeMap<String, String>>::from_headers(message.headers());
        sqlx::query("INSERT INTO export_jobs (meta, payload) VALUES (?, ?)")
            .bind(headers)
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[subscriber(InboxQueue::<ExportJob>::new("exports"))]
async fn send_export(export: &Export, Headers(tenant): Headers<Tenant>) -> HandlerOutcome {
    if tenant.tenant != "acme" {
        tracing::warn!(tenant = %tenant.tenant, "a tenant this service does not serve");
        return HandlerOutcome::drop();
    }
    tracing::info!(export.report, tenant = %tenant.tenant, "exporting");
    HandlerOutcome::ack()
}

examples/manual/sqlx_headers.rs
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Headers as HeadersColumn, Lease, Payload, ProcessedAt};
use ruststream_sqlx::{HeaderColumn, HeaderRow, InboxSpec, InboxTable, PayloadRow};
use sqlx::types::Json;

#[derive(sqlx::FromRow)]
struct ExportJob {
    id: i64,
    meta: Option<Json<BTreeMap<String, String>>>,
    payload: Vec<u8>,
}

impl InboxTable for ExportJob {
    type Id = i64;
    type Table = InboxSpec<(
        ProcessedAt<DateTime<Utc>>,
        Lease<DateTime<Utc>>,
        HeadersColumn,
        Payload,
    )>;
    const TABLE: Self::Table = InboxSpec::new("export_jobs", Column::new("id").generated())
        .processed_at(Column::new("processed_at"))
        .lease(Column::new("locked_until"))
        .headers(Column::new("meta"))
        .payload(Column::new("payload"));

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

impl HeaderRow for ExportJob {
    type Column = Option<Json<BTreeMap<String, String>>>;

    fn headers_mut(&mut self) -> &mut Self::Column {
        &mut self.meta
    }
}

impl PayloadRow for ExportJob {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

impl Publish<Sqlite> for ExportJob {
    async fn publish(
        conn: &mut SqliteConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        // The column keeps the message's headers as the delivery reads them back.
        let headers = Json::<BTreeMap<String, String>>::from_headers(message.headers());
        sqlx::query("INSERT INTO export_jobs (meta, payload) VALUES (?, ?)")
            .bind(headers)
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

struct Exporter;

impl Handle<Export> for Exporter {
    fn handle(
        &self,
        export: &Export,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let tenant = ctx.headers().get_str("x-tenant").unwrap_or_default();
        if tenant != "acme" {
            tracing::warn!(tenant, "a tenant this service does not serve");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(export.report, tenant, "exporting");
        ready(Ok(()))
    }
}

Example​

Example