Trait HeaderColumn 

pub trait HeaderColumn {
    // Required methods
    fn take_headers(&mut self) -> HeaderMap;
    fn from_headers(headers: &HeaderMap) -> Self;
    fn unfit(headers: &HeaderMap) -> Option<&str>;
}

The type of a field playing headers: what the delivery’s header map holds.

sqlx::types::Json<BTreeMap<String, String>> implements it under the json feature (a jsonb or json column of string values), and so does an Option of any implementation. A publish carrying a header the column cannot hold byte for byte is refused before the service’s Publish runs: a header never reaches the table changed.

Examples

use ruststream::HeaderMap;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::HeaderColumn;
use serde::Deserialize;
use sqlx::PgPool;

/// Headers stored one per line, `name: value`, in a `TEXT` column.
#[derive(sqlx::Type)]
#[sqlx(transparent)]
pub 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 = "jobs")]
pub struct Job {
    #[field(id, generated)]
    id: i64,
    #[field(headers)]
    headers: Lines,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Task {
    n: u32,
}

// The delivery's headers come from `headers`, so the service's middleware reads `x-tenant`
// there as on any broker.
#[subscriber(InboxQueue::<Job>::new("tasks"))]
async fn run(task: &Task) -> HandlerOutcome {
    tracing::info!(task.n, "running");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(run);
    })
}

Required Methods

fn take_headers(&mut self) -> HeaderMap

Moves the headers the column holds into a header map, leaving the column empty.

A delivery takes them once, when its row is claimed, so a column that owns its strings hands them over without a copy.

fn from_headers(headers: &HeaderMap) -> Self

The column value that holds headers, for a service’s Publish.

Examples

use std::collections::BTreeMap;

use ruststream::OutgoingMessage;
use ruststream_sqlx::{HeaderColumn, Inbox, Publish};
use sqlx::types::Json;
use sqlx::{PgConnection, Postgres};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "jobs")]
pub struct Job {
    #[field(id, generated)]
    id: i64,
    #[field(headers)]
    headers: Json<BTreeMap<String, String>>,
    #[field(payload)]
    payload: Vec<u8>,
}

impl Publish<Postgres> for Job {
    async fn publish(
        conn: &mut PgConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        // The `jsonb` column holds the message's headers as the delivery will read them back.
        let headers = Json::<BTreeMap<String, String>>::from_headers(message.headers());
        sqlx::query("INSERT INTO jobs (headers, payload) VALUES ($1, $2)")
            .bind(headers)
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

fn unfit(headers: &HeaderMap) -> Option<&str>

The first of headers the column cannot hold byte for byte, if any: the broker refuses a publish that carries it.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementations on Foreign Types

impl HeaderColumn for Json<BTreeMap<String, String>>

Available on crate feature json only.

A JSON object of strings. A value that is not UTF-8 does not fit: the broker refuses a publish that carries one, so the replacement from_headers would write never reaches the table.

fn take_headers(&mut self) -> HeaderMap

fn from_headers(headers: &HeaderMap) -> Self

fn unfit(headers: &HeaderMap) -> Option<&str>

impl<T: HeaderColumn> HeaderColumn for Option<T>

fn take_headers(&mut self) -> HeaderMap

fn from_headers(headers: &HeaderMap) -> Self

fn unfit(headers: &HeaderMap) -> Option<&str>

Implementors