ruststream_sqlx_dialect/advisory.rs
1//! The advisory lock form: a lock on each row's key, held by the delivery's session, holds the row.
2
3use crate::dialect::Dialect;
4use crate::spec::TableSpec;
5use crate::statement::{ClaimShape, Statement, StatementError};
6
7/// The advisory lock form of a dialect: the claim of candidates, the lock and the unlock of a key in
8/// the session, and the take of a candidate once its lock is held.
9///
10/// A table with `#[inbox(advisory_lock = "..")]` takes its rows this way. The key template names
11/// columns of the row ([`Form::Advisory`](crate::Form::Advisory)), and the database renders each
12/// row's key from them as text. A claim selects candidates with their keys
13/// ([`advisory_claim`](Self::advisory_claim)), locks each key on the delivery's own connection
14/// ([`lock`](Self::lock), bound as [`Param::Key`](crate::Param::Key)), and takes the row while it
15/// is still claimable ([`take`](Self::take)), counting its attempt. No transaction stays open
16/// while the handler runs. The settlements are the statements every form runs, naming the row
17/// alone, and the unlock ([`unlock`](Self::unlock)) follows them; a crash ends the session and
18/// releases its locks at once. A dialect implements this trait where it builds these statements,
19/// and a table with `advisory_lock` does not compile against a dialect that does not. Every
20/// built-in dialect implements it.
21///
22/// The lock and the unlock return one 64-bit integer, nonzero when the lock was taken or released,
23/// so one decode serves every database. A dialect whose database keeps no locks returns neither
24/// statement, and the broker keeps the keys in work in the process: [`Sqlite`](crate::Sqlite) does.
25///
26/// # Examples
27///
28/// A dialect for a database without a built-in one builds the advisory lock form itself. On SQL
29/// Server an application lock the delivery's session owns holds the row:
30///
31/// ```
32/// # use std::num::NonZeroUsize;
33/// # use ruststream_sqlx_dialect::TableName;
34/// use ruststream_sqlx_dialect::{
35/// Advisory, ClaimShape, Dialect, Param, Statement, StatementError, TableSpec,
36/// };
37///
38/// /// SQL Server, a database without a built-in dialect.
39/// #[derive(Debug)]
40/// pub struct Mssql;
41///
42/// // A table with `advisory_lock` mounts on `Mssql`.
43/// impl Advisory for Mssql {
44/// // The candidates of `email_jobs` (`advisory_lock = "email-{job_id}"`) in id order, each
45/// // with its key. `APPLOCK_TEST` leaves out the keys another session holds.
46/// fn advisory_claim(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
47/// Ok(Statement::new(
48/// "SELECT TOP (@p1) [job_id], CONCAT(N'email-', [job_id]) AS [__lock] \
49/// FROM [email_jobs] \
50/// WHERE APPLOCK_TEST('public', CONCAT(N'email-', [job_id]), \
51/// 'Exclusive', 'Session') = 1 \
52/// ORDER BY [job_id]",
53/// [Param::Limit],
54/// ))
55/// }
56///
57/// // An application lock the session owns, tried without waiting: `sp_getapplock` answers
58/// // zero or more when it granted the lock.
59/// fn lock(&self) -> Option<Statement> {
60/// Some(Statement::new(
61/// "DECLARE @result int; \
62/// EXEC @result = sp_getapplock @Resource = @p1, @LockMode = 'Exclusive', \
63/// @LockOwner = 'Session', @LockTimeout = 0; \
64/// SELECT CAST(CASE WHEN @result >= 0 THEN 1 ELSE 0 END AS bigint)",
65/// [Param::Key],
66/// ))
67/// }
68///
69/// // The release of the session's lock. A key the session does not hold answers zero, as
70/// // `sp_releaseapplock` would raise an error for it.
71/// fn unlock(&self) -> Option<Statement> {
72/// Some(Statement::new(
73/// "IF APPLOCK_MODE('public', @p1, 'Session') = 'NoLock' SELECT CAST(0 AS bigint) \
74/// ELSE BEGIN DECLARE @result int; \
75/// EXEC @result = sp_releaseapplock @Resource = @p1, @LockOwner = 'Session'; \
76/// SELECT CAST(CASE WHEN @result = 0 THEN 1 ELSE 0 END AS bigint) END",
77/// [Param::Key],
78/// ))
79/// }
80///
81/// fn take(
82/// &self,
83/// _spec: &TableSpec<'_>,
84/// _shape: ClaimShape,
85/// ) -> Result<Vec<Statement>, StatementError> {
86/// Ok(vec![Statement::new(
87/// "UPDATE [email_jobs] SET [attempt] = [attempt] + 1 \
88/// OUTPUT deleted.[job_id], deleted.[attempt], deleted.[payload] \
89/// WHERE [job_id] = @p1",
90/// [Param::Id],
91/// )])
92/// }
93/// }
94/// # impl Dialect for Mssql {
95/// # fn name(&self) -> &'static str { "mssql" }
96/// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
97/// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
98/// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
99/// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
100/// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
101/// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
102/// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
103/// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
104/// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
105/// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
106/// # }
107/// ```
108#[diagnostic::on_unimplemented(
109 message = "the `{Self}` dialect takes no advisory locks, so a table on it cannot take its rows \
110 by advisory lock",
111 label = "a table with `advisory_lock` takes its rows by advisory lock",
112 note = "declare `#[field(locked_until)]` (the lease form), or drop `advisory_lock` (the row lock \
113 form)",
114 note = "a dialect of the service's own takes the advisory lock form by implementing \
115 `Advisory`: the claim of candidates, the lock, the unlock and the take"
116)]
117pub trait Advisory: Dialect {
118 /// The candidates: up to [`Param::Limit`](crate::Param::Limit) claimable rows of the
119 /// subscription's group in claim order, two columns each, the id and the lock key as text
120 /// (named `__lock`). It takes no lock and writes nothing; where the database can tell, it
121 /// leaves out the rows whose key another session holds.
122 ///
123 /// # Errors
124 ///
125 /// [`StatementError::FormMismatch`] for a table in another form;
126 /// [`StatementError::AdvisoryFifo`] for a table with FIFO groups, as the lock key keeps a
127 /// group in order in this form.
128 ///
129 /// # Examples
130 ///
131 /// ```
132 /// # use std::num::NonZeroUsize;
133 /// # use ruststream_sqlx_dialect::TableName;
134 /// use ruststream_sqlx_dialect::{
135 /// Advisory, ClaimShape, Dialect, Param, Statement, StatementError, TableSpec,
136 /// };
137 ///
138 /// /// SQL Server, a database without a built-in dialect.
139 /// #[derive(Debug)]
140 /// pub struct Mssql;
141 ///
142 /// impl Advisory for Mssql {
143 /// // The candidates of `email_jobs` (`advisory_lock = "email-{job_id}"`) in id order, each
144 /// // with its key. `APPLOCK_TEST` leaves out the keys another session holds.
145 /// fn advisory_claim(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
146 /// Ok(Statement::new(
147 /// "SELECT TOP (@p1) [job_id], CONCAT(N'email-', [job_id]) AS [__lock] \
148 /// FROM [email_jobs] \
149 /// WHERE APPLOCK_TEST('public', CONCAT(N'email-', [job_id]), \
150 /// 'Exclusive', 'Session') = 1 \
151 /// ORDER BY [job_id]",
152 /// [Param::Limit],
153 /// ))
154 /// }
155 /// # fn lock(&self) -> Option<Statement> { Some(Statement::new("DECLARE @result int; EXEC @result = sp_getapplock @Resource = @p1, @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0; SELECT CAST(CASE WHEN @result >= 0 THEN 1 ELSE 0 END AS bigint)", [Param::Key])) }
156 /// # fn unlock(&self) -> Option<Statement> { Some(Statement::new("IF APPLOCK_MODE('public', @p1, 'Session') = 'NoLock' SELECT CAST(0 AS bigint) ELSE BEGIN DECLARE @result int; EXEC @result = sp_releaseapplock @Resource = @p1, @LockOwner = 'Session'; SELECT CAST(CASE WHEN @result = 0 THEN 1 ELSE 0 END AS bigint) END", [Param::Key])) }
157 /// # fn take(&self, _spec: &TableSpec<'_>, _shape: ClaimShape) -> Result<Vec<Statement>, StatementError> { Ok(vec![Statement::new("UPDATE [email_jobs] SET [attempt] = [attempt] + 1 OUTPUT deleted.[job_id], deleted.[attempt], deleted.[payload] WHERE [job_id] = @p1", [Param::Id])]) }
158 /// }
159 /// # impl Dialect for Mssql {
160 /// # fn name(&self) -> &'static str { "mssql" }
161 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
162 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
163 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
164 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
165 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
166 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
167 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
168 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
169 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
170 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
171 /// # }
172 /// ```
173 fn advisory_claim(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError>;
174
175 /// Tries the lock on [`Param::Key`](crate::Param::Key) for the session, without waiting: one
176 /// row whose first column is a 64-bit integer, nonzero when the lock was taken. `None` when the
177 /// process keeps the locks.
178 ///
179 /// # Examples
180 ///
181 /// ```
182 /// # use std::num::NonZeroUsize;
183 /// # use ruststream_sqlx_dialect::TableName;
184 /// use ruststream_sqlx_dialect::{
185 /// Advisory, ClaimShape, Dialect, Param, Statement, StatementError, TableSpec,
186 /// };
187 ///
188 /// /// SQL Server, a database without a built-in dialect.
189 /// #[derive(Debug)]
190 /// pub struct Mssql;
191 ///
192 /// impl Advisory for Mssql {
193 /// // An application lock the session owns, tried without waiting: `sp_getapplock` answers
194 /// // zero or more when it granted the lock.
195 /// fn lock(&self) -> Option<Statement> {
196 /// Some(Statement::new(
197 /// "DECLARE @result int; \
198 /// EXEC @result = sp_getapplock @Resource = @p1, @LockMode = 'Exclusive', \
199 /// @LockOwner = 'Session', @LockTimeout = 0; \
200 /// SELECT CAST(CASE WHEN @result >= 0 THEN 1 ELSE 0 END AS bigint)",
201 /// [Param::Key],
202 /// ))
203 /// }
204 /// # fn advisory_claim(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Ok(Statement::new("SELECT TOP (@p1) [job_id], CONCAT(N'email-', [job_id]) AS [__lock] FROM [email_jobs] WHERE APPLOCK_TEST('public', CONCAT(N'email-', [job_id]), 'Exclusive', 'Session') = 1 ORDER BY [job_id]", [Param::Limit])) }
205 /// # fn unlock(&self) -> Option<Statement> { Some(Statement::new("IF APPLOCK_MODE('public', @p1, 'Session') = 'NoLock' SELECT CAST(0 AS bigint) ELSE BEGIN DECLARE @result int; EXEC @result = sp_releaseapplock @Resource = @p1, @LockOwner = 'Session'; SELECT CAST(CASE WHEN @result = 0 THEN 1 ELSE 0 END AS bigint) END", [Param::Key])) }
206 /// # fn take(&self, _spec: &TableSpec<'_>, _shape: ClaimShape) -> Result<Vec<Statement>, StatementError> { Ok(vec![Statement::new("UPDATE [email_jobs] SET [attempt] = [attempt] + 1 OUTPUT deleted.[job_id], deleted.[attempt], deleted.[payload] WHERE [job_id] = @p1", [Param::Id])]) }
207 /// }
208 /// # impl Dialect for Mssql {
209 /// # fn name(&self) -> &'static str { "mssql" }
210 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
211 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
212 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
213 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
214 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
215 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
216 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
217 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
218 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
219 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
220 /// # }
221 /// ```
222 fn lock(&self) -> Option<Statement>;
223
224 /// Releases the lock on [`Param::Key`](crate::Param::Key): one row whose first column is a
225 /// 64-bit integer, nonzero when the session held it. `None` when the process keeps the locks.
226 ///
227 /// # Examples
228 ///
229 /// ```
230 /// # use std::num::NonZeroUsize;
231 /// # use ruststream_sqlx_dialect::TableName;
232 /// use ruststream_sqlx_dialect::{
233 /// Advisory, ClaimShape, Dialect, Param, Statement, StatementError, TableSpec,
234 /// };
235 ///
236 /// /// SQL Server, a database without a built-in dialect.
237 /// #[derive(Debug)]
238 /// pub struct Mssql;
239 ///
240 /// impl Advisory for Mssql {
241 /// // The release of the session's lock. A key the session does not hold answers zero, as
242 /// // `sp_releaseapplock` would raise an error for it.
243 /// fn unlock(&self) -> Option<Statement> {
244 /// Some(Statement::new(
245 /// "IF APPLOCK_MODE('public', @p1, 'Session') = 'NoLock' SELECT CAST(0 AS bigint) \
246 /// ELSE BEGIN DECLARE @result int; \
247 /// EXEC @result = sp_releaseapplock @Resource = @p1, @LockOwner = 'Session'; \
248 /// SELECT CAST(CASE WHEN @result = 0 THEN 1 ELSE 0 END AS bigint) END",
249 /// [Param::Key],
250 /// ))
251 /// }
252 /// # fn advisory_claim(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Ok(Statement::new("SELECT TOP (@p1) [job_id], CONCAT(N'email-', [job_id]) AS [__lock] FROM [email_jobs] WHERE APPLOCK_TEST('public', CONCAT(N'email-', [job_id]), 'Exclusive', 'Session') = 1 ORDER BY [job_id]", [Param::Limit])) }
253 /// # fn lock(&self) -> Option<Statement> { Some(Statement::new("DECLARE @result int; EXEC @result = sp_getapplock @Resource = @p1, @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0; SELECT CAST(CASE WHEN @result >= 0 THEN 1 ELSE 0 END AS bigint)", [Param::Key])) }
254 /// # fn take(&self, _spec: &TableSpec<'_>, _shape: ClaimShape) -> Result<Vec<Statement>, StatementError> { Ok(vec![Statement::new("UPDATE [email_jobs] SET [attempt] = [attempt] + 1 OUTPUT deleted.[job_id], deleted.[attempt], deleted.[payload] WHERE [job_id] = @p1", [Param::Id])]) }
255 /// }
256 /// # impl Dialect for Mssql {
257 /// # fn name(&self) -> &'static str { "mssql" }
258 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
259 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
260 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
261 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
262 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
263 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
264 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
265 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
266 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
267 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
268 /// # }
269 /// ```
270 fn unlock(&self) -> Option<Statement>;
271
272 /// The statements that take a candidate whose lock the session holds: they count its attempt
273 /// where the table has one and read the row again in `shape`, as it was before the count, only
274 /// while it is still claimable. The last returns the row, or none when the row is gone or no
275 /// longer claimable; a first statement, where there are two, counts the attempt and changes no row
276 /// in that case.
277 ///
278 /// A table read with `*` ([`TableSpec::selects_all`]) names no column, so its row returns
279 /// with the attempt counted.
280 ///
281 /// # Errors
282 ///
283 /// [`StatementError::FormMismatch`] for a table in another form;
284 /// [`StatementError::AdvisoryFifo`] for a table with FIFO groups.
285 ///
286 /// # Examples
287 ///
288 /// ```
289 /// # use std::num::NonZeroUsize;
290 /// # use ruststream_sqlx_dialect::TableName;
291 /// use ruststream_sqlx_dialect::{
292 /// Advisory, ClaimShape, Dialect, Param, Statement, StatementError, TableSpec,
293 /// };
294 ///
295 /// /// SQL Server, a database without a built-in dialect.
296 /// #[derive(Debug)]
297 /// pub struct Mssql;
298 ///
299 /// impl Advisory for Mssql {
300 /// // Taking an email also records when. `email_jobs` keeps no finished rows and no delays,
301 /// // so a row that is still there is claimable; `deleted` returns it as it was before the
302 /// // count.
303 /// fn take(
304 /// &self,
305 /// _spec: &TableSpec<'_>,
306 /// shape: ClaimShape,
307 /// ) -> Result<Vec<Statement>, StatementError> {
308 /// let returned = match shape {
309 /// ClaimShape::Rows => "deleted.[job_id], deleted.[attempt], deleted.[payload]",
310 /// ClaimShape::Ids => "deleted.[job_id]",
311 /// ClaimShape::Roles => {
312 /// "deleted.[job_id] AS [id], deleted.[attempt] AS [attempt], \
313 /// deleted.[payload] AS [payload]"
314 /// }
315 /// };
316 /// Ok(vec![Statement::new(
317 /// format!(
318 /// "UPDATE [email_jobs] SET [attempt] = [attempt] + 1, [taken_at] = @p1 \
319 /// OUTPUT {returned} WHERE [job_id] = @p2"
320 /// ),
321 /// [Param::Now, Param::Id],
322 /// )])
323 /// }
324 /// # fn advisory_claim(&self, _spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Ok(Statement::new("SELECT TOP (@p1) [job_id], CONCAT(N'email-', [job_id]) AS [__lock] FROM [email_jobs] WHERE APPLOCK_TEST('public', CONCAT(N'email-', [job_id]), 'Exclusive', 'Session') = 1 ORDER BY [job_id]", [Param::Limit])) }
325 /// # fn lock(&self) -> Option<Statement> { Some(Statement::new("DECLARE @result int; EXEC @result = sp_getapplock @Resource = @p1, @LockMode = 'Exclusive', @LockOwner = 'Session', @LockTimeout = 0; SELECT CAST(CASE WHEN @result >= 0 THEN 1 ELSE 0 END AS bigint)", [Param::Key])) }
326 /// # fn unlock(&self) -> Option<Statement> { Some(Statement::new("IF APPLOCK_MODE('public', @p1, 'Session') = 'NoLock' SELECT CAST(0 AS bigint) ELSE BEGIN DECLARE @result int; EXEC @result = sp_releaseapplock @Resource = @p1, @LockOwner = 'Session'; SELECT CAST(CASE WHEN @result = 0 THEN 1 ELSE 0 END AS bigint) END", [Param::Key])) }
327 /// }
328 /// # impl Dialect for Mssql {
329 /// # fn name(&self) -> &'static str { "mssql" }
330 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
331 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
332 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
333 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
334 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
335 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
336 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
337 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
338 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
339 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
340 /// # }
341 /// ```
342 fn take(
343 &self,
344 spec: &TableSpec<'_>,
345 shape: ClaimShape,
346 ) -> Result<Vec<Statement>, StatementError>;
347}