Events

custom(..) hands one event of a table to the service's own SQL, Claim to Extend, and every other event keeps the statement the dialect builds.

An event the table names, in #[inbox(custom(..))] on the derive or with .own::<own::X>() by hand, runs the service's own code in place of the dialect's statement. Each event is a trait the struct implements for its database:

Event	Trait	What it does
`claim`	`Claim`	selects and takes the ids of the next rows
`fetch`	`Fetch`	reads the claimed rows, such as over a join
`ack`	`Ack`	finishes a row its handler acknowledged
`retry`	`Retry`	returns a row to the queue at once
`retry_after`	`RetryAfter`	hides a row until a time
`discard`	`Discard`	finishes a row its handler dropped
`dead_letter`	`DeadLetter`	moves a row past its cap
`extend`	`Extend`	extends the leases in work
`lock`, `unlock`	`Lock`, `Unlock`	take and release the key of a row

A named event without its impl does not compile, and the error names the trait. The table and the broker stay as they were, and every event the table does not name keeps the dialect's statement.

A settlement of the service's own runs in the transaction the crate commits, so it settles the row as the built-in one does. An acknowledgement takes the row out of what the claim selects: a row it leaves claimable is delivered again. The form limits the events: the advisory lock form selects its candidates by key itself and takes no claim, extend belongs to the lease form, and lock and unlock come together in the advisory lock form.

Example

A sent email stays in its table for an audit: the acknowledgement moves it to the sent group instead of deleting it.

Macros
Manual

examples/sqlx_tables_events.rs
use ruststream_sqlx::Ack;
use sqlx::{Sqlite, SqliteConnection};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs", custom(ack))]
struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

// A sent email stays in its table, in the `sent` group, for an audit.
impl Ack<Sqlite> for SendEmail {
    async fn ack(conn: &mut SqliteConnection, id: &i64) -> Result<(), sqlx::Error> {
        sqlx::query("UPDATE email_jobs SET name = 'sent' WHERE job_id = ?")
            .bind(id)
            .execute(conn)
            .await?;
        Ok(())
    }
}

examples/manual/sqlx_tables_events.rs
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Lease, Payload, own};
use ruststream_sqlx::{Ack, InboxSpec, InboxTable, PayloadRow};
use sqlx::{Sqlite, SqliteConnection};

#[derive(sqlx::FromRow)]
struct SendEmail {
    job_id: i64,
    payload: Vec<u8>,
}

impl InboxTable for SendEmail {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, Payload, own::Ack)>;
    const TABLE: Self::Table = InboxSpec::new("email_jobs", Column::new("job_id").generated())
        .group(Column::new("name"))
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"))
        .own::<own::Ack>();

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

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

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

// A sent email stays in its table, in the `sent` group, for an audit.
impl Ack<Sqlite> for SendEmail {
    async fn ack(conn: &mut SqliteConnection, id: &i64) -> Result<(), sqlx::Error> {
        sqlx::query("UPDATE email_jobs SET name = 'sent' WHERE job_id = ?")
            .bind(id)
            .execute(conn)
            .await?;
        Ok(())
    }
}

Example​

Example