Inbox

#[derive(Inbox)] describes a queue table with an ordinary sqlx struct: the table in #[inbox(..)], column roles in #[field(..)], names from sqlx.

The struct derives sqlx::FromRow beside Inbox, and the queue is a table of the service's own schema. #[inbox(table = "..")] names the table and is required; schema = ".." places it in a schema, and neither holds a dot. #[field(id)] marks the row's identity, the one role every table has; its type is Clone. The other roles, such as group, attempt and locked_until, each turn on one behaviour of the queue (Roles). sqlx's own attributes name the columns. A struct with a #[field(payload)] field is in payload mode, and one without it is in row mode.

The rest of #[inbox(..)] selects how the table is served:

Attribute	What it does
`advisory_lock = "jobs-{id}"`	the advisory lock form, with a key built from the fields in braces
`isolation = ..`, `mode = ..`	the isolation level or the SQLite mode the table's transactions open at
`clock = ..`	where "now" comes from
`custom(..)`	the events the service writes itself
`checked, db = ..`	the statements in sqlx's compile-time check

A generic struct keeps its parameters: the impl asks Send + Sync + 'static of the struct and Clone + Debug + Send + Sync + 'static of the id field's type. A struct that cannot drive a queue does not compile, and the error points at the field or the name: no id, a role played twice, a column named twice, a dot in table, isolation beside mode, and the form rules of Events. The derive writes the InboxTable impl a table described by hand writes, and the traits its roles need.

Example

One table holds the notifications of every channel: its channel column is the group, and each subscription reads the group its name selects.

Macros
Manual

examples/sqlx_tables_inbox.rs
use chrono::{DateTime, Utc};
use sqlx::Sqlite;

/// One row of `notifications`: the struct names the table, sqlx names the columns, and
/// `#[field(..)]` gives the columns that run the queue their roles.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "notifications")]
struct Notification {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    channel: String,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<Notification>::new("email"))]
async fn email(note: &Note, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    tracing::info!(to = %note.to, ?attempt, "emailing");
    HandlerOutcome::ack()
}

#[subscriber(InboxQueue::<Notification>::new("sms"))]
async fn sms(note: &Note) -> HandlerOutcome {
    tracing::info!(to = %note.to, "texting");
    HandlerOutcome::ack()
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    let broker = broker
        .route::<Notification>("email")
        .route::<Notification>("sms");
    RustStream::new(AppInfo::new("notifier", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            b.include(email);
            b.include(sms);
        })
}

examples/manual/sqlx_tables_inbox.rs
use chrono::{DateTime, Utc};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Attempt, Lease, Payload};
use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable, PayloadRow};
use sqlx::Sqlite;

/// One row of `notifications`, described by hand: the columns only the queue reads, such as
/// `channel` and `locked_until`, need no field.
#[derive(sqlx::FromRow)]
struct Notification {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for Notification {
    type Id = i64;
    type Table = InboxSpec<(Attempt, Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("notifications", Column::new("id").generated())
        .group(Column::new("channel"))
        .attempt(Column::new("attempt").generated())
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

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

impl AttemptRow for Notification {
    type Attempt = i16;

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

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

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

struct Deliver;

impl Handle<Note> for Deliver {
    fn handle(
        &self,
        note: &Note,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(to = %note.to, channel = ctx.name(), "delivering");
        ready(Ok(()))
    }
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    let broker = broker
        .route::<Notification>("email")
        .route::<Notification>("sms");
    RustStream::new(AppInfo::new("notifier", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            for channel in ["email", "sms"] {
                let deliver = subscriber(channel, Deliver)
                    .map_source(move |_| InboxQueue::<Notification>::new(channel));
                b.include(deliver.build());
            }
        })
}

Example​

Example