Derive Macro InboxHeaders 

#[derive(InboxHeaders)]
{
    // Attributes available to this derive:
    #[inbox]
    #[field]
    #[sqlx]
}

Describes the queue table of a message assembled from it, and implements InboxHeaders for the struct.

The struct takes the table’s attributes and the fields that play a role, as a struct deriving Inbox does; every field without a role is a header (see InboxHeaders). custom(..), payload and headers belong to the message struct, which flattens this one into its #[field(headers)] field. #[inbox(checked, db = ..)] checks the table’s statements at compile time, and the message then reads its rows in a Fetch of its own (the headers layout, checked).

Examples

use ruststream_sqlx::prelude::*;
use sqlx::PgPool;

#[derive(Debug, Clone, InboxHeaders, sqlx::FromRow)]
#[inbox(table = "order_jobs")]
pub struct OrderHeaders {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    tenant: String,
}

#[derive(Debug, Clone, Inbox, sqlx::FromRow)]
pub struct OrderJob {
    #[field(headers)]
    #[sqlx(flatten)]
    headers: OrderHeaders,
    note: Option<String>,
}

#[subscriber(InboxQueue::<OrderJob>::new("orders"))]
async fn ship(job: &OrderJob) -> HandlerOutcome {
    tracing::info!(tenant = %job.headers.tenant, note = ?job.note, "shipping");
    HandlerOutcome::ack()
}

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

With checked on the headers struct, the message lists fetch in custom(..) and reads its rows in a Fetch of its own, where sqlx::query! checks that statement too:

use chrono::{DateTime, Utc};
use ruststream_sqlx::Fetch;
use ruststream_sqlx::prelude::*;
use sqlx::{Error, FromRow, PgConnection, Postgres};

// order_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL, processed_at TIMESTAMPTZ,
// trace TEXT, total BIGINT NOT NULL
#[derive(Debug, Clone, InboxHeaders, FromRow)]
#[inbox(table = "order_jobs", checked, db = postgres)]
pub struct OrderHeaders {
    #[field(id, generated)]
    pub job_id: i64,
    #[field(group)]
    pub name: String,
    #[field(processed_at)]
    pub processed_at: Option<DateTime<Utc>>,
    pub trace: Option<String>,
}

#[derive(Debug, Clone, Inbox, FromRow)]
#[inbox(custom(fetch))]
pub struct Order {
    #[field(headers)]
    #[sqlx(flatten)]
    pub headers: OrderHeaders,
    pub total: i64,
}

impl Fetch<Postgres> for Order {
    async fn fetch(conn: &mut PgConnection, ids: &[i64]) -> Result<Vec<Self>, Error> {
        let rows = sqlx::query!(
            "SELECT job_id, name, processed_at, trace, total FROM order_jobs \
             WHERE job_id = ANY($1)",
            ids
        )
        .fetch_all(conn)
        .await?;
        Ok(rows
            .into_iter()
            .map(|row| Self {
                headers: OrderHeaders {
                    job_id: row.job_id,
                    name: row.name,
                    processed_at: row.processed_at,
                    trace: row.trace,
                },
                total: row.total,
            })
            .collect())
    }
}

Implemented in ruststream-sqlx-macros and used through ruststream-sqlx.