跳到主要内容
本页尚未翻译

下面显示的是英文原文。

Dialect

SqlxBroker::with_dialect serves a database with a dialect of the service's own: Dialect, then the trait of each form it builds, and ByName for names.

A dialect of the service's own serves a database without a built-in dialect, such as CockroachDB through sqlx's Postgres driver. Its type is the broker's second type parameter, SqlxBroker<Postgres, Cockroach>, and each thing its tables can do is a trait it implements:

  • Dialect: names, placeholders, the statement that opens a transaction at a table's level, the savepoint a transactional handler's writes start after, and the statements every form runs to settle a row, move a dead letter, fetch rows and insert one;
  • RowLock: the row lock form and its claim;
  • Lease: the lease form, its claim, the extension of a lease in work and the stamp of a row a claim only selected;
  • Advisory: the advisory lock form, its claim of candidates with their keys, the lock and the release of a key, and the take of a row whose key the session holds;
  • Opens<Level>: an isolation level or SQLite mode a table may declare;
  • ByName<DB>: subscriptions by name, the JSON headers and the times their rows hold.

A table in a form whose trait the dialect lacks does not compile, and neither does a by-name mount on a dialect without ByName; the error names the trait. A dialect refuses with a StatementError what it does not build, such as FIFO groups, and the subscription of such a table stops at startup. The provided methods of Dialect and Lease state what their defaults mean for a database without the feature. The statements are built once, when a subscription opens, so a message costs the same as through a built-in dialect.

A dialect may wrap a built-in one, such as dialect::Sqlite, and write one statement its own way. It then forwards every other method, the provided methods the built-in dialect overrides included, or the wrapped statements change. One statement written the service's own way is usually an event of the table instead, which needs no dialect. The crate registers the brokers of its built-in dialects with the test harness; a test of a broker on a dialect of the service's own registers its type once, with ruststream::register_testable_broker!(SqlxBroker<Sqlite, Audited>), and addresses it by that type, tb.broker::<SqlxBroker<Sqlite, Audited>>().

Example​

Audited wraps the SQLite dialect and acknowledges a row by moving it to the group acknowledged, where it stays for an audit.

examples/sqlx_databases_dialect.rs
/// One row of `email_jobs`, claimed under a lease.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
struct SendEmail {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

use ruststream_sqlx::dialect::{
    self, ClaimShape, Dialect, Lease, Opening, Param, Role, Statement, StatementError, TableName,
    TableSpec,
};

/// The group an acknowledged row moves to.
const ACKNOWLEDGED: &str = "acknowledged";

/// SQLite, with an acknowledgement of the service's own: an acknowledged row of a table with
/// groups moves to the `acknowledged` group and stays there for an audit.
#[derive(Debug)]
struct Audited;

impl Dialect for Audited {
    fn name(&self) -> &'static str {
        "audited"
    }

    fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        let Some(group) = spec.column(Role::Group) else {
            return dialect::Sqlite.ack(spec);
        };
        let mut sql = String::from("UPDATE ");
        self.quote_into(spec.table(), &mut sql);
        sql.push_str(" SET ");
        self.quote_into(group.name(), &mut sql);
        sql.push_str(" = '");
        sql.push_str(ACKNOWLEDGED);
        sql.push_str("' WHERE ");
        self.quote_into(spec.id().name(), &mut sql);
        sql.push_str(" = ");
        self.placeholder_into(NonZeroUsize::MIN, &mut sql);
        Ok(Statement::new(sql, [Param::Id]))
    }

    // Every other statement is the built-in dialect's, the provided methods it overrides included.
    fn quote_into(&self, ident: &str, out: &mut String) {
        dialect::Sqlite.quote_into(ident, out);
    }

    fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) {
        dialect::Sqlite.placeholder_into(index, out);
    }

    fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        dialect::Sqlite.fetch(spec)
    }

    fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> {
        dialect::Sqlite.retry(spec)
    }

    fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        dialect::Sqlite.retry_after(spec)
    }

    fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        dialect::Sqlite.discard(spec)
    }

    fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        dialect::Sqlite.dead_letter_group(spec)
    }

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

    fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        dialect::Sqlite.insert(spec)
    }

    fn begin(&self, opening: Opening) -> Result<Option<&'static str>, StatementError> {
        dialect::Sqlite.begin(opening)
    }
}

// The lease form, which `SendEmail` takes: its claim, the extension of a lease in work and the
// stamp of a claimed row.
impl Lease for Audited {
    fn lease_claim(
        &self,
        spec: &TableSpec<'_>,
        shape: ClaimShape,
    ) -> Result<Statement, StatementError> {
        dialect::Sqlite.lease_claim(spec, shape)
    }

    fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        dialect::Sqlite.extend(spec)
    }

    fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        dialect::Sqlite.stamp(spec)
    }

    fn claim_counts_attempt(&self, spec: &TableSpec<'_>) -> bool {
        dialect::Sqlite.claim_counts_attempt(spec)
    }

    fn begin_lease_claim(&self) -> Option<&'static str> {
        dialect::Sqlite.begin_lease_claim()
    }
}

// A `SqlxBroker<Sqlite, Audited>`: the dialect is part of the broker's type.
fn broker(pool: SqlitePool) -> SqlxBroker<Sqlite, Audited> {
    SqlxBroker::with_dialect(pool, Audited).route::<SendEmail>("emails")
}