ruststream_sqlx_dialect/form.rs
1//! How a subscription claims the rows of a table: the forms, and the pieces of an advisory
2//! lock key.
3
4use crate::column::Column;
5
6/// One piece of an advisory lock key: literal text, or the value of a column of the row.
7///
8/// `#[inbox(advisory_lock = "jobs-{job_id}")]` becomes
9/// `[KeyPart::Literal("jobs-"), KeyPart::Column("job_id")]`. The database renders each row's key
10/// from the parts, in the claim that selects the row: a literal as it is, a column's value as
11/// text, a column without a value as empty text.
12///
13/// # Examples
14///
15/// ```
16/// # use std::num::NonZeroUsize;
17/// # use ruststream_sqlx_dialect::TableName;
18/// use ruststream_sqlx_dialect::{
19/// Advisory, ClaimShape, Dialect, Form, KeyPart, Param, Statement, StatementError, TableSpec,
20/// };
21///
22/// /// SQL Server, a database without a built-in dialect; the service's advisory tables keep no
23/// /// groups, no claim order, no delays and no finished rows.
24/// #[derive(Debug)]
25/// pub struct Mssql;
26///
27/// impl Advisory for Mssql {
28/// // The candidates of a table read in id order, with their keys.
29/// fn advisory_claim(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
30/// let Form::Advisory(key) = spec.form() else {
31/// return Err(StatementError::FormMismatch {
32/// statement: "advisory_claim",
33/// form: spec.form().name(),
34/// });
35/// };
36/// // `advisory_lock = "jobs-{job_id}"` arrives as `[Literal("jobs-"), Column("job_id")]`,
37/// // and the database renders each row's key from it. `CONCAT` reads a column without a
38/// // value as empty text, and the leading empty literal gives it the two arguments it
39/// // takes at least.
40/// let mut rendered = String::from("N''");
41/// for part in key {
42/// rendered.push_str(", ");
43/// match part {
44/// KeyPart::Literal(text) => {
45/// rendered.push_str("N'");
46/// rendered.push_str(&text.replace('\'', "''"));
47/// rendered.push('\'');
48/// }
49/// KeyPart::Column(column) => self.quote_into(column, &mut rendered),
50/// }
51/// }
52/// let mut table = String::new();
53/// self.quote_into(spec.table(), &mut table);
54/// let mut id = String::new();
55/// self.quote_into(spec.id().name(), &mut id);
56/// Ok(Statement::new(
57/// format!(
58/// "SELECT TOP (@p1) {id}, CONCAT({rendered}) AS [__lock] \
59/// FROM {table} ORDER BY {id}"
60/// ),
61/// [Param::Limit],
62/// ))
63/// }
64/// # fn lock(&self) -> Option<Statement> { None }
65/// # fn unlock(&self) -> Option<Statement> { None }
66/// # fn take(&self, spec: &TableSpec<'_>, shape: ClaimShape) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
67/// }
68/// # impl Dialect for Mssql {
69/// # fn name(&self) -> &'static str { "mssql" }
70/// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
71/// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
72/// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
73/// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
74/// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
75/// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
76/// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
77/// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
78/// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
79/// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
80/// # }
81/// ```
82#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
83pub enum KeyPart<'a> {
84 /// Text copied into the key as it is.
85 Literal(&'a str),
86 /// The value of the named column.
87 Column(&'a str),
88}
89
90/// How a subscription claims the rows of a table; a table has exactly one form.
91///
92/// The lease form carries its `locked_until` column and the advisory form its key, so a table
93/// cannot declare two forms. The row lock is the default: the form of a table that declares
94/// neither.
95///
96/// # Examples
97///
98/// A settlement of a dialect of the service's own follows the form:
99///
100/// ```
101/// # use std::num::NonZeroUsize;
102/// # use ruststream_sqlx_dialect::TableName;
103/// use ruststream_sqlx_dialect::{Dialect, Form, Param, Statement, StatementError, TableSpec};
104///
105/// /// SQL Server, a database without a built-in dialect.
106/// #[derive(Debug)]
107/// pub struct Mssql;
108///
109/// impl Dialect for Mssql {
110/// fn name(&self) -> &'static str {
111/// "mssql"
112/// }
113///
114/// fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
115/// let mut sql = String::from("DELETE FROM ");
116/// self.quote_into(spec.table(), &mut sql);
117/// sql.push_str(" WHERE ");
118/// self.quote_into(spec.id().name(), &mut sql);
119/// sql.push_str(" = @p1");
120/// match spec.form() {
121/// // In the lease form the row goes only while it holds the delivery's lease.
122/// Form::Lease(expiry) => {
123/// sql.push_str(" AND ");
124/// self.quote_into(expiry.name(), &mut sql);
125/// sql.push_str(" = @p2");
126/// Ok(Statement::new(sql, [Param::Id, Param::Held]))
127/// }
128/// _ => Ok(Statement::new(sql, [Param::Id])),
129/// }
130/// }
131/// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
132/// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
133/// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
134/// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
135/// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
136/// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
137/// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
138/// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
139/// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
140/// }
141/// ```
142#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
143#[non_exhaustive]
144pub enum Form<'a> {
145 /// Rows stay locked in a transaction held for the whole handler; the default.
146 #[default]
147 RowLock,
148 /// A claim sets this column, the lease's expiry, and commits; the value holds the row.
149 Lease(Column<'a>),
150 /// A session lock on the key these parts build from the row holds the row.
151 Advisory(&'a [KeyPart<'a>]),
152}
153
154impl Form<'_> {
155 /// The form's name in messages: `row lock`, `lease` or `advisory lock`.
156 ///
157 /// # Examples
158 ///
159 /// ```
160 /// # use std::num::NonZeroUsize;
161 /// # use ruststream_sqlx_dialect::TableName;
162 /// use ruststream_sqlx_dialect::{Dialect, Form, Param, Statement, StatementError, TableSpec};
163 ///
164 /// /// SQL Server, a database without a built-in dialect.
165 /// #[derive(Debug)]
166 /// pub struct Mssql;
167 ///
168 /// impl Dialect for Mssql {
169 /// fn name(&self) -> &'static str {
170 /// "mssql"
171 /// }
172 ///
173 /// fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
174 /// // This dialect serves the row lock form alone, and its refusal names the table's
175 /// // form.
176 /// if spec.form() != Form::RowLock {
177 /// return Err(StatementError::UnsupportedForm {
178 /// dialect: self.name(),
179 /// form: spec.form().name(),
180 /// });
181 /// }
182 /// let mut sql = String::from("DELETE FROM ");
183 /// self.quote_into(spec.table(), &mut sql);
184 /// sql.push_str(" WHERE ");
185 /// self.quote_into(spec.id().name(), &mut sql);
186 /// sql.push_str(" = @p1");
187 /// Ok(Statement::new(sql, [Param::Id]))
188 /// }
189 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
190 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
191 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
192 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
193 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
194 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
195 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
196 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
197 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
198 /// }
199 /// ```
200 #[must_use]
201 pub const fn name(&self) -> &'static str {
202 match self {
203 Self::RowLock => "row lock",
204 Self::Lease(_) => "lease",
205 Self::Advisory(_) => "advisory lock",
206 }
207 }
208}
209
210#[cfg(test)]
211mod tests {
212 use super::{Column, Form};
213
214 #[test]
215 fn form_names_read_in_messages() {
216 assert_eq!(Form::RowLock.name(), "row lock");
217 assert_eq!(Form::Lease(Column::new("locked_until")).name(), "lease");
218 assert_eq!(Form::Advisory(&[]).name(), "advisory lock");
219 }
220
221 #[test]
222 fn the_row_lock_is_the_default_form() {
223 assert_eq!(Form::default(), Form::RowLock);
224 }
225}