ruststream_sqlx_dialect/role.rs
1//! The roles a column plays for the queue, as `#[field(..)]` names them.
2
3use std::fmt::{self, Display, Formatter};
4
5/// What a column does for the queue, named as `#[field(..)]` spells it.
6///
7/// A struct marks the columns that run its queue with roles; every other column is the
8/// message's own data.
9///
10/// # Examples
11///
12/// ```
13/// # use std::num::NonZeroUsize;
14/// # use ruststream_sqlx_dialect::TableName;
15/// use ruststream_sqlx_dialect::{Dialect, Param, Role, Statement, StatementError, TableSpec};
16///
17/// /// SQL Server, a database without a built-in dialect.
18/// #[derive(Debug)]
19/// pub struct Mssql;
20///
21/// impl Dialect for Mssql {
22/// fn name(&self) -> &'static str {
23/// "mssql"
24/// }
25///
26/// fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
27/// // A delayed retry writes the column that plays `retry_after`.
28/// let Some(until) = spec.column(Role::RetryAfter) else {
29/// return Err(StatementError::MissingRole {
30/// statement: "retry_after",
31/// role: Role::RetryAfter,
32/// });
33/// };
34/// let mut sql = String::from("UPDATE ");
35/// self.quote_into(spec.table(), &mut sql);
36/// sql.push_str(" SET ");
37/// self.quote_into(until.name(), &mut sql);
38/// sql.push_str(" = @p1 WHERE ");
39/// self.quote_into(spec.id().name(), &mut sql);
40/// sql.push_str(" = @p2");
41/// Ok(Statement::new(sql, [Param::RetryAfter, Param::Id]))
42/// }
43/// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
44/// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
45/// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
46/// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
47/// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
48/// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
49/// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
50/// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
51/// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
52/// }
53/// ```
54#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
55#[non_exhaustive]
56pub enum Role {
57 /// The row's identity, for settlement and for matching fetched rows to claimed ids.
58 Id,
59 /// The group a subscription reads: the rows of one table split into groups.
60 Group,
61 /// The key of the delivery's lane under `workers(n, by_key)` or `threads(n, by_key)`.
62 PartitionKey,
63 /// The claim order: a smaller value is claimed first.
64 Priority,
65 /// The time before which a row is not claimed.
66 RetryAfter,
67 /// The attempt number; the first delivery reads 1.
68 Attempt,
69 /// The expiry of a lease; a column playing it selects the lease form.
70 LockedUntil,
71 /// The time a row was finished; acknowledgement sets it instead of deleting the row.
72 ProcessedAt,
73 /// The delivery's headers.
74 Headers,
75 /// The message bytes a handler decodes.
76 Payload,
77}
78
79impl Role {
80 /// Every role, in the order the documentation lists them.
81 // The derives and the crate read it to parse and check a table; a service never names it.
82 #[doc(hidden)]
83 pub const ALL: &'static [Self] = &[
84 Self::Id,
85 Self::Group,
86 Self::PartitionKey,
87 Self::Priority,
88 Self::RetryAfter,
89 Self::Attempt,
90 Self::LockedUntil,
91 Self::ProcessedAt,
92 Self::Headers,
93 Self::Payload,
94 ];
95
96 /// The role's name inside `#[field(..)]`.
97 ///
98 /// # Examples
99 ///
100 /// ```
101 /// # use std::num::NonZeroUsize;
102 /// use ruststream_sqlx_dialect::{
103 /// ClaimShape, Dialect, Param, Role, RowLock, Statement, StatementError, TableSpec,
104 /// };
105 /// # use ruststream_sqlx_dialect::TableName;
106 ///
107 /// /// SQL Server, a database without a built-in dialect.
108 /// #[derive(Debug)]
109 /// pub struct Mssql;
110 ///
111 /// impl RowLock for Mssql {
112 /// // The service's queue tables have no groups and no claim order of their own.
113 /// fn lock_claim(
114 /// &self,
115 /// spec: &TableSpec<'_>,
116 /// shape: ClaimShape,
117 /// ) -> Result<Statement, StatementError> {
118 /// let mut selected = String::new();
119 /// match shape {
120 /// // Whole rows, for a struct the broker reads.
121 /// ClaimShape::Rows => {
122 /// for column in spec.columns() {
123 /// if !selected.is_empty() {
124 /// selected.push_str(", ");
125 /// }
126 /// self.quote_into(column.name(), &mut selected);
127 /// }
128 /// }
129 /// // Only the ids, for a struct that fetches its rows itself.
130 /// ClaimShape::Ids => self.quote_into(spec.id().name(), &mut selected),
131 /// // The columns that run the queue, each under the name of its role.
132 /// ClaimShape::Roles => {
133 /// let roles = [
134 /// Role::Id,
135 /// Role::PartitionKey,
136 /// Role::Attempt,
137 /// Role::Headers,
138 /// Role::Payload,
139 /// ];
140 /// for role in roles {
141 /// let Some(column) = spec.column(role) else {
142 /// continue;
143 /// };
144 /// if !selected.is_empty() {
145 /// selected.push_str(", ");
146 /// }
147 /// self.quote_into(column.name(), &mut selected);
148 /// selected.push_str(" AS ");
149 /// self.quote_into(role.attribute(), &mut selected);
150 /// }
151 /// }
152 /// }
153 /// let mut sql = format!("SELECT TOP (@p1) {selected} FROM ");
154 /// self.quote_into(spec.table(), &mut sql);
155 /// sql.push_str(" WITH (UPDLOCK, READPAST) ORDER BY ");
156 /// self.quote_into(spec.id().name(), &mut sql);
157 /// Ok(Statement::new(sql, [Param::Limit]))
158 /// }
159 /// }
160 /// # impl Dialect for Mssql {
161 /// # fn name(&self) -> &'static str { "mssql" }
162 /// # fn quote_into(&self, ident: &str, out: &mut String) { out.push('['); out.push_str(&ident.replace(']', "]]")); out.push(']'); }
163 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
164 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
165 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
166 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
167 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
168 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
169 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
170 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
171 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
172 /// # }
173 /// ```
174 #[must_use]
175 pub const fn attribute(self) -> &'static str {
176 match self {
177 Self::Id => "id",
178 Self::Group => "group",
179 Self::PartitionKey => "partition_key",
180 Self::Priority => "priority",
181 Self::RetryAfter => "retry_after",
182 Self::Attempt => "attempt",
183 Self::LockedUntil => "locked_until",
184 Self::ProcessedAt => "processed_at",
185 Self::Headers => "headers",
186 Self::Payload => "payload",
187 }
188 }
189
190 /// The role `#[field(..)]` names with `name`, or `None` when no role has that name.
191 // The derives and the crate read it to parse and check a table; a service never names it.
192 #[doc(hidden)]
193 #[must_use]
194 pub fn from_attribute(name: &str) -> Option<Self> {
195 Self::ALL
196 .iter()
197 .copied()
198 .find(|role| role.attribute() == name)
199 }
200}
201
202impl Display for Role {
203 fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
204 f.write_str(self.attribute())
205 }
206}
207
208#[cfg(test)]
209mod tests {
210 use super::Role;
211
212 #[test]
213 fn every_role_reads_back_from_its_attribute() {
214 for role in Role::ALL {
215 assert_eq!(Role::from_attribute(role.attribute()), Some(*role));
216 assert_eq!(role.to_string(), role.attribute());
217 }
218 assert_eq!(Role::ALL.len(), 10);
219 assert_eq!(Role::from_attribute("deadline"), None);
220 }
221}