Lock

custom(lock, unlock) hands the lock and the release of each key to the service's SQL, so a queue shares a lock another system already takes.

The struct implements Lock and Unlock for its database. lock tries the key without waiting and answers whether it took it. unlock answers whether the session held the key; an answer of false, or an error, closes the connection, which ends the session and its locks. The two come together, and only in the advisory lock form; a description that breaks either rule does not compile. The dialect still selects the candidates and takes each row, so the claim keeps its order and counts the attempt as before. The process keeps no registry of keys for such a table, on SQLite too: the service's lock decides. A claim reads the first due rows alone and takes nothing while their keys are held, so the next row waits up to a poll interval. On Postgres a lock of the service's own is the place for pg_try_advisory_lock on a key another system computes; on SQLite it may be a table of the keys in work, as here. By hand, the chain adds .own::<own::Lock>() and .own::<own::Unlock>(), and the description's type lists own::Lock and own::Unlock beside Advisory. The other events of a table are on Events.

Example

Billing holds an account's key in held_keys while it changes the account, and a payout of that account waits until billing releases the key.

Macros
Manual

examples/sqlx_forms_lock.rs
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::{Lock, Unlock};
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `payouts`, held by the key of its account.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(
    table = "payouts",
    advisory_lock = "accounts-{account_id}",
    custom(lock, unlock)
)]
struct Payout {
    #[field(id, generated)]
    id: i64,
    account_id: i64,
    #[field(payload)]
    payload: Vec<u8>,
}

/// Billing holds an account's key in `held_keys` while it changes the account, so a payout
/// never runs beside a billing run of its account.
impl Lock<Sqlite> for Payout {
    async fn lock(conn: &mut SqliteConnection, key: &str) -> Result<bool, sqlx::Error> {
        let taken = sqlx::query("INSERT OR IGNORE INTO held_keys (key) VALUES (?)")
            .bind(key)
            .execute(conn)
            .await?;
        Ok(taken.rows_affected() == 1)
    }
}

impl Unlock<Sqlite> for Payout {
    async fn unlock(conn: &mut SqliteConnection, key: &str) -> Result<bool, sqlx::Error> {
        let released = sqlx::query("DELETE FROM held_keys WHERE key = ?")
            .bind(key)
            .execute(conn)
            .await?;
        Ok(released.rows_affected() == 1)
    }
}

#[subscriber(InboxQueue::<Payout>::new("payouts"))]
async fn pay(transfer: &Transfer) -> HandlerOutcome {
    tracing::info!(transfer.cents, "paying out");
    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());
    RustStream::new(AppInfo::new("payouts", "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(pay);
        })
}

examples/manual/sqlx_forms_lock.rs
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::{Lock, Unlock};
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `payouts`, held by the key of its account.
#[derive(sqlx::FromRow)]
struct Payout {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Payout {
    type Id = i64;
    type Table = InboxSpec<(Advisory, Payload, own::Lock, own::Unlock)>;
    const TABLE: Self::Table = InboxSpec::new("payouts", Column::new("id").generated())
        .advisory(&[KeyPart::Literal("accounts-"), KeyPart::Column("account_id")])
        .payload(Column::new("payload"))
        .own::<own::Lock>()
        .own::<own::Unlock>();

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

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

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

/// Billing holds an account's key in `held_keys` while it changes the account, so a payout
/// never runs beside a billing run of its account.
impl Lock<Sqlite> for Payout {
    async fn lock(conn: &mut SqliteConnection, key: &str) -> Result<bool, sqlx::Error> {
        let taken = sqlx::query("INSERT OR IGNORE INTO held_keys (key) VALUES (?)")
            .bind(key)
            .execute(conn)
            .await?;
        Ok(taken.rows_affected() == 1)
    }
}

impl Unlock<Sqlite> for Payout {
    async fn unlock(conn: &mut SqliteConnection, key: &str) -> Result<bool, sqlx::Error> {
        let released = sqlx::query("DELETE FROM held_keys WHERE key = ?")
            .bind(key)
            .execute(conn)
            .await?;
        Ok(released.rows_affected() == 1)
    }
}

struct Pay;

impl Handle<Transfer> for Pay {
    fn handle(
        &self,
        transfer: &Transfer,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(transfer.cents, "paying out");
        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());
    RustStream::new(AppInfo::new("payouts", "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| {
            let pay =
                subscriber("payouts", Pay).map_source(|_| InboxQueue::<Payout>::new("payouts"));
            b.include(pay.build());
        })
}

Example​

Example