HeaderColumn

HeaderColumn lets a headers column hold a type of the service's own; it moves the map in and out and names a header it cannot keep byte for byte.

The trait has three methods. take_headers moves the headers the column holds into a header map, once per delivery as its row is claimed, so a column that owns its strings hands them over without a copy. from_headers builds the column value that holds a message's headers, for the service's Publish. unfit names the first header the column cannot hold byte for byte, and the broker refuses a publish that carries it with SqlxBrokerError::Header.

The crate implements it for Json<BTreeMap<String, String>> under the json feature, and for Option<T> of any implementation, where None holds no headers. The trait is shared by the inbox and the outbox: one column type serves a queue table and an outbox table alike. The type is also a sqlx type of its own, so the driver reads and writes the column through it.

Example

Lines keeps the headers one per line, name: value, in a TEXT column, and refuses a value that is not text on one line.

Macros

examples/sqlx_headers_column.rs

use ruststream::HeaderMap;
use ruststream_sqlx::HeaderColumn;

/// Headers kept one per line, `name: value`, in a `TEXT` column.
#[derive(sqlx::Type)]
#[sqlx(transparent)]
struct Lines(String);

impl HeaderColumn for Lines {
    fn take_headers(&mut self) -> HeaderMap {
        let headers = self
            .0
            .lines()
            .filter_map(|line| line.split_once(": "))
            .map(|(name, value)| (name.to_owned(), value.to_owned()))
            .collect();
        self.0.clear();
        headers
    }

    fn from_headers(headers: &HeaderMap) -> Self {
        let lines: Vec<String> = headers
            .iter()
            .map(|(name, value)| format!("{name}: {}", String::from_utf8_lossy(value)))
            .collect();
        Self(lines.join("\n"))
    }

    fn unfit(headers: &HeaderMap) -> Option<&str> {
        // A value comes back as it went in when it is text on one line.
        let fits = |value: &[u8]| str::from_utf8(value).is_ok_and(|text| !text.contains('\n'));
        headers
            .iter()
            .find(|(_, value)| !fits(value))
            .map(|(name, _)| name)
    }
}

#[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: Lines,
    #[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 = Lines::from_headers(message.headers());
        sqlx::query("INSERT INTO export_jobs (meta, payload) VALUES (?, ?)")
            .bind(headers)
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

Manual

examples/manual/sqlx_headers_column.rs

use ruststream::HeaderMap;
use ruststream_sqlx::HeaderColumn;

/// Headers kept one per line, `name: value`, in a `TEXT` column.
#[derive(sqlx::Type)]
#[sqlx(transparent)]
struct Lines(String);

impl HeaderColumn for Lines {
    fn take_headers(&mut self) -> HeaderMap {
        let headers = self
            .0
            .lines()
            .filter_map(|line| line.split_once(": "))
            .map(|(name, value)| (name.to_owned(), value.to_owned()))
            .collect();
        self.0.clear();
        headers
    }

    fn from_headers(headers: &HeaderMap) -> Self {
        let lines: Vec<String> = headers
            .iter()
            .map(|(name, value)| format!("{name}: {}", String::from_utf8_lossy(value)))
            .collect();
        Self(lines.join("\n"))
    }

    fn unfit(headers: &HeaderMap) -> Option<&str> {
        // A value comes back as it went in when it is text on one line.
        let fits = |value: &[u8]| str::from_utf8(value).is_ok_and(|text| !text.contains('\n'));
        headers
            .iter()
            .find(|(_, value)| !fits(value))
            .map(|(name, _)| name)
    }
}

#[derive(sqlx::FromRow)]
struct ExportJob {
    id: i64,
    meta: Lines,
    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 = Lines;

    fn headers_mut(&mut self) -> &mut Lines {
        &mut self.meta
    }
}

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 = Lines::from_headers(message.headers());
        sqlx::query("INSERT INTO export_jobs (meta, payload) VALUES (?, ?)")
            .bind(headers)
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}