Struct BuiltIn 

pub struct BuiltIn<DB: BuiltInDialect>(/* private fields */);

The dialect built into the crate for the database DB: what SqlxBroker::new builds its statements with.

It builds them with the dialect of the crate’s dialect module that serves the database: Postgres on Postgres, MySql on MySQL and MariaDB, Sqlite on SQLite. On an AnyPool it is the dialect of the database the pool reaches, among those whose features are on, picked when the broker connects.

It serves the forms its database serves, through the traits of the dialect module: every BuiltIn implements Lease and Advisory, and BuiltIn<Postgres>, BuiltIn<MySql> and BuiltIn<Any> implement RowLock. A SQLite table in the row lock form therefore does not compile. An AnyPool names its database only when the broker connects, so on a SQLite backend the subscription to a row lock table stops when it starts, with StatementError::UnsupportedForm.

It opens transactions at the isolation levels and SQLite modes its database keeps, one Opens per level: BuiltIn<Postgres> READ COMMITTED, REPEATABLE READ and SERIALIZABLE, BuiltIn<MySql> those and READ UNCOMMITTED, BuiltIn<Sqlite> the three modes. A table that names another does not compile. BuiltIn<Any> takes all seven, and the picked backend’s dialect refuses the ones its database lacks when the subscription starts, with StatementError::UnsupportedOpening.

Examples

use ruststream_sqlx::BuiltIn;
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::{PgPool, Postgres};

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

#[derive(Deserialize)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

// `SqlxBroker<Postgres>` names the same type: the built-in dialect is the default.
pub fn broker(pool: PgPool) -> SqlxBroker<Postgres, BuiltIn<Postgres>> {
    SqlxBroker::new(pool)
}

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

Trait Implementations

impl<DB: BuiltInDialect> Advisory for BuiltIn<DB>

fn advisory_claim( &self, spec: &TableSpec<'_>, ) -> Result<Statement, StatementError>

The candidates: up to Param::Limit claimable rows of the subscription’s group in claim order, two columns each, the id and the lock key as text (named __lock). It takes no lock and writes nothing; where the database can tell, it leaves out the rows whose key another session holds.

fn lock(&self) -> Option<Statement>

Tries the lock on Param::Key for the session, without waiting: one row whose first column is a 64-bit integer, nonzero when the lock was taken. None when the process keeps the locks.

fn unlock(&self) -> Option<Statement>

Releases the lock on Param::Key: one row whose first column is a 64-bit integer, nonzero when the session held it. None when the process keeps the locks.

fn take( &self, spec: &TableSpec<'_>, shape: ClaimShape, ) -> Result<Vec<Statement>, StatementError>

The statements that take a candidate whose lock the session holds: they count its attempt where the table has one and read the row again in shape, as it was before the count, only while it is still claimable. The last returns the row, or none when the row is gone or no longer claimable; a first statement, where there are two, counts the attempt and changes no row in that case.

impl ByName<Any> for BuiltIn<Any>

Available on crate feature any only.

fn headers( value: <Any as Database>::ValueRef<'_>, ) -> Result<HeaderMap, BoxDynError>

The headers a row keeps as a JSON object of strings.

fn bind_time( arguments: &mut <Any as Database>::Arguments, time: NamedTime, ) -> Result<(), Error>

Binds time: the current time, a delayed retry’s, or a lease’s expiry.

impl ByName<MySql> for BuiltIn<MySql>

fn headers( value: <MySql as Database>::ValueRef<'_>, ) -> Result<HeaderMap, BoxDynError>

The headers a row keeps as a JSON object of strings.

fn bind_time( arguments: &mut <MySql as Database>::Arguments, time: NamedTime, ) -> Result<(), Error>

Binds time: the current time, a delayed retry’s, or a lease’s expiry.

impl ByName<Postgres> for BuiltIn<Postgres>

fn headers( value: <Postgres as Database>::ValueRef<'_>, ) -> Result<HeaderMap, BoxDynError>

The headers a row keeps as a JSON object of strings.

fn bind_time( arguments: &mut <Postgres as Database>::Arguments, time: NamedTime, ) -> Result<(), Error>

Binds time: the current time, a delayed retry’s, or a lease’s expiry.

impl ByName<Sqlite> for BuiltIn<Sqlite>

fn headers( value: <Sqlite as Database>::ValueRef<'_>, ) -> Result<HeaderMap, BoxDynError>

The headers a row keeps as a JSON object of strings.

fn bind_time( arguments: &mut <Sqlite as Database>::Arguments, time: NamedTime, ) -> Result<(), Error>

Binds time: the current time, a delayed retry’s, or a lease’s expiry.

impl<DB: BuiltInDialect> Clone for BuiltIn<DB>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<DB: BuiltInDialect> Copy for BuiltIn<DB>

impl<DB: BuiltInDialect> Debug for BuiltIn<DB>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<DB: BuiltInDialect> Dialect for BuiltIn<DB>

fn name(&self) -> &'static str

The dialect’s name, for messages.

fn quote_into(&self, ident: &str, out: &mut String)

Appends ident to out, quoted as a name of this database.

fn placeholder_into(&self, index: NonZeroUsize, out: &mut String)

Appends the placeholder of parameter number index (counted from 1) to out.

fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that reads the rows of claimed ids, bound as one list.

fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that acknowledges a row: it deletes the row, or sets processed_at when the table has that column.

fn retry( &self, spec: &TableSpec<'_>, ) -> Result<Option<Statement>, StatementError>

The statement that releases a row for another attempt at once, or None when releasing the row needs no statement.

fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that releases a row for another attempt after a delay, bound as Param::RetryAfter.

fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that drops a row: it deletes the row, or sets processed_at when the table has that column.

fn dead_letter_group( &self, spec: &TableSpec<'_>, ) -> Result<Statement, StatementError>

The statement that moves a row whose attempts are spent to another group, bound as Param::Destination.

fn dead_letter_table( &self, spec: &TableSpec<'_>, target: TableName<'_>, ) -> Result<Vec<Statement>, StatementError>

The statements that move a row whose attempts are spent to target, a table with the same columns; they run in one transaction.

fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that inserts a row: every column the database does not fill, in the order of TableSpec::columns, each bound as Param::Column.

fn server_version(&self) -> Option<&'static str>

The query that reads the server’s version as one text column, or None when the dialect’s statements run on every version of its server.

fn check_server( &self, spec: &TableSpec<'_>, version: &str, ) -> Result<(), StatementError>

Refuses a server older than the statements of spec need; version is what the server_version query returned.

fn begin( &self, opening: Opening, ) -> Result<Option<&'static str>, StatementError>

The statement that opens a transaction at opening in place of BEGIN, or None when BEGIN opens it.

fn savepoint(&self) -> &'static str

The statement that sets the savepoint after a claim, inside the claim’s transaction: what a handler writes after it can be discarded while the claim’s own work stays.

fn rollback_to_savepoint(&self) -> &'static str

The statement that discards what the transaction did after savepoint, and keeps the transaction open.

fn fifo_guard( &self, spec: &TableSpec<'_>, ) -> Result<Option<Statement>, StatementError>

The statement a claim of a table with FIFO groups runs first in its transaction, to take the subscription’s group for it; None for a table without FIFO groups, or a dialect that takes no group.

impl<DB: BuiltInDialect> Lease for BuiltIn<DB>

fn lease_claim( &self, spec: &TableSpec<'_>, shape: ClaimShape, ) -> Result<Statement, StatementError>

The statement that claims up to Param::Limit rows of the subscription’s group, in claim order, among the rows whose lease ended by Param::LeaseNow.

fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that extends a delivery’s lease: it writes the new expiry (Param::Lease) while the row still holds the delivery’s token (Param::Held), so a handler that runs longer than one lease keeps its row.

fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>

The statement that leases one claimed row: it writes the expiry (Param::Lease) and counts the attempt, while no lease holds the row at Param::LeaseNow.

fn claim_writes_lease(&self) -> bool

Whether the lease claim writes the lease itself. A dialect whose lease claim only selects the rows answers false, and the claim’s transaction then runs stamp for each claimed row before it commits.

fn claim_counts_attempt(&self, spec: &TableSpec<'_>) -> bool

Whether the rows the lease claim of spec returns carry the attempt the claim counted. A dialect whose claim returns the rows as they were before answers false; where it answers true, a delivery reports one less than its row carries.

fn begin_lease_claim(&self) -> Option<&'static str>

The statement that opens the transaction of a claim that only selects its rows, in place of BEGIN, or None when BEGIN opens it.

impl Opens<Deferred> for BuiltIn<Sqlite>

Available on crate feature sqlite only.

impl Opens<Deferred> for BuiltIn<Any>

Available on crate feature any only.

impl Opens<Exclusive> for BuiltIn<Sqlite>

Available on crate feature sqlite only.

impl Opens<Exclusive> for BuiltIn<Any>

Available on crate feature any only.

impl Opens<Immediate> for BuiltIn<Sqlite>

Available on crate feature sqlite only.

impl Opens<Immediate> for BuiltIn<Any>

Available on crate feature any only.

impl Opens<ReadCommitted> for BuiltIn<Postgres>

Available on crate feature postgres only.

impl Opens<ReadCommitted> for BuiltIn<MySql>

Available on crate feature mysql only.

impl Opens<ReadCommitted> for BuiltIn<Any>

Available on crate feature any only.

impl Opens<ReadUncommitted> for BuiltIn<MySql>

Available on crate feature mysql only.

impl Opens<ReadUncommitted> for BuiltIn<Any>

Available on crate feature any only.

impl Opens<RepeatableRead> for BuiltIn<Postgres>

Available on crate feature postgres only.

impl Opens<RepeatableRead> for BuiltIn<MySql>

Available on crate feature mysql only.

impl Opens<RepeatableRead> for BuiltIn<Any>

Available on crate feature any only.

impl Opens<Serializable> for BuiltIn<Postgres>

Available on crate feature postgres only.

impl Opens<Serializable> for BuiltIn<MySql>

Available on crate feature mysql only.

impl Opens<Serializable> for BuiltIn<Any>

Available on crate feature any only.

impl RowLock for BuiltIn<Postgres>

Available on crate feature postgres only.

fn lock_claim( &self, spec: &TableSpec<'_>, shape: ClaimShape, ) -> Result<Statement, StatementError>

The statement that claims up to Param::Limit rows of the subscription’s group, in claim order, and locks them until the claim’s transaction ends.

impl RowLock for BuiltIn<MySql>

Available on crate feature mysql only.

fn lock_claim( &self, spec: &TableSpec<'_>, shape: ClaimShape, ) -> Result<Statement, StatementError>

The statement that claims up to Param::Limit rows of the subscription’s group, in claim order, and locks them until the claim’s transaction ends.

impl RowLock for BuiltIn<Any>

Available on crate feature any only.

fn lock_claim( &self, spec: &TableSpec<'_>, shape: ClaimShape, ) -> Result<Statement, StatementError>

The statement that claims up to Param::Limit rows of the subscription’s group, in claim order, and locks them until the claim’s transaction ends.