Struct Statement

Source

pub struct Statement { /* private fields */ }

Expand description

One SQL statement: its text and the parameters its placeholders bind, in placeholder order.

§Examples

use ruststream_sqlx_dialect::{Dialect, Param, Statement, StatementError, TableSpec};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

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

    // A dropped email stays in its table, marked with the time it was dropped.
    fn discard(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
        Ok(Statement::new(
            "UPDATE [email_jobs] SET [dropped_at] = @p1 WHERE [job_id] = @p2",
            [Param::Now, Param::Id],
        ))
    }
}

Implementations§

Source §

impl Statement

Source

pub fn new( sql: impl Into<String>, params: impl IntoIterator<Item = Param>, ) -> Self

A statement with its text and the parameters its placeholders bind, in order.

§Examples

use ruststream_sqlx_dialect::{
    ClaimShape, Dialect, Param, RowLock, Statement, StatementError, TableSpec,
};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl RowLock for Mssql {
    // The claim of the service's one queue table, which it reads whole.
    fn lock_claim(
        &self,
        spec: &TableSpec<'_>,
        _shape: ClaimShape,
    ) -> Result<Statement, StatementError> {
        // This claim keeps no group in order.
        if spec.is_fifo() {
            return Err(StatementError::UnsupportedFifo { dialect: self.name() });
        }
        Ok(Statement::new(
            "SELECT TOP (@p1) [job_id], [payload] FROM [email_jobs] \
             WITH (UPDLOCK, READPAST) ORDER BY [job_id]",
            [Param::Limit],
        ))
    }
}

Source

pub fn sql(&self) -> &str

The statement’s text.

§Examples

use ruststream_sqlx_dialect::{
    ClaimShape, Dialect, Lease, Param, Statement, StatementError, TableSpec,
};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl Mssql {
    // The claimable rows of `email_jobs`, which both the lease claim and a claim of the
    // service's own (`custom(claim)`) read.
    fn candidates(&self) -> Statement {
        Statement::new(
            "SELECT TOP (@p2) * FROM [email_jobs] WITH (UPDLOCK, READPAST, ROWLOCK) \
             WHERE [locked_until] IS NULL OR [locked_until] <= @p1 ORDER BY [job_id]",
            [Param::LeaseNow, Param::Limit],
        )
    }
}

impl Lease for Mssql {
    // The lease claim updates the candidates in place: their text becomes the claim's
    // common table expression.
    fn lease_claim(
        &self,
        _spec: &TableSpec<'_>,
        _shape: ClaimShape,
    ) -> Result<Statement, StatementError> {
        let candidates = self.candidates();
        Ok(Statement::new(
            format!(
                "WITH [claimed] AS ({}) UPDATE [claimed] \
                 SET [locked_until] = @p3, [attempt] = [attempt] + 1 \
                 OUTPUT deleted.[job_id], deleted.[attempt], deleted.[payload]",
                candidates.sql()
            ),
            [Param::LeaseNow, Param::Limit, Param::Lease],
        ))
    }
}

Source

pub fn params(&self) -> &[Param]

The parameters the placeholders bind, in placeholder order.

§Examples

use ruststream_sqlx_dialect::{
    ClaimShape, Dialect, Lease, Param, Statement, StatementError, TableSpec,
};

/// SQL Server, a database without a built-in dialect.
#[derive(Debug)]
pub struct Mssql;

impl Mssql {
    // The claimable rows of `email_jobs`, which both the lease claim and a claim of the
    // service's own (`custom(claim)`) read.
    fn candidates(&self) -> Statement {
        Statement::new(
            "SELECT TOP (@p2) * FROM [email_jobs] WITH (UPDLOCK, READPAST, ROWLOCK) \
             WHERE [locked_until] IS NULL OR [locked_until] <= @p1 ORDER BY [job_id]",
            [Param::LeaseNow, Param::Limit],
        )
    }
}

impl Lease for Mssql {
    // The lease claim updates the candidates in place. Its parameters are theirs, then the
    // lease it writes, so `@p3` binds the lease.
    fn lease_claim(
        &self,
        _spec: &TableSpec<'_>,
        _shape: ClaimShape,
    ) -> Result<Statement, StatementError> {
        let candidates = self.candidates();
        Ok(Statement::new(
            format!(
                "WITH [claimed] AS ({}) UPDATE [claimed] \
                 SET [locked_until] = @p3, [attempt] = [attempt] + 1 \
                 OUTPUT deleted.[job_id], deleted.[attempt], deleted.[payload]",
                candidates.sql()
            ),
            candidates.params().iter().copied().chain([Param::Lease]),
        ))
    }
}