ruststream_sqlx_dialect/
spec.rs

1//! The description of a queue table: its columns, the role each one plays, and how rows are
2//! claimed.
3
4mod builder;
5mod reading;
6
7use crate::column::Column;
8use crate::form::Form;
9use crate::opening::Opening;
10
11// The `const` inserts walk the columns with the same walker as `TableSpec::columns`.
12#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
13pub(crate) use reading::Columns;
14
15/// Whether the rows of a table split into groups, and whether each group keeps its order.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
17enum Grouping<'a> {
18    None,
19    Groups(Column<'a>),
20    Fifo(Column<'a>),
21}
22
23/// The description of a queue table: its name, the column that identifies a row, one slot per
24/// role, the message's data columns, the form its rows are claimed in, and what its transactions
25/// open at.
26///
27/// `#[derive(Inbox)]` builds one as a constant, and a dialect reads it to build statements. The id
28/// column is part of the constructor and every role has one slot, which its setter fills once (a
29/// second call panics), so a table without an id or with a role played twice cannot be described.
30/// Column names are strings, so the types do not catch a name used twice or a lock key reading a
31/// column the table lacks: `#[derive(Inbox)]` refuses both at compile time, and a description
32/// written by hand meets the database's own checks when the statements are prepared.
33///
34/// # Examples
35///
36/// A dialect of the service's own reads the description to build a statement:
37///
38/// ```
39/// # use std::num::NonZeroUsize;
40/// use ruststream_sqlx_dialect::{
41///     Dialect, Form, Param, Role, Statement, StatementError, TableSpec,
42/// };
43/// # use ruststream_sqlx_dialect::TableName;
44///
45/// /// SQL Server, a database without a built-in dialect, keeping the finished rows of a table with
46/// /// groups in its `done` group.
47/// #[derive(Debug)]
48/// pub struct Mssql;
49///
50/// impl Dialect for Mssql {
51///     fn name(&self) -> &'static str {
52///         "mssql"
53///     }
54///
55///     fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
56///         // The service's tables take their rows by row lock.
57///         if spec.form() != Form::RowLock {
58///             return Err(StatementError::UnsupportedForm {
59///                 dialect: self.name(),
60///                 form: spec.form().name(),
61///             });
62///         }
63///         // A table that names no schema lives in SQL Server's default one, `dbo`.
64///         let mut table = String::new();
65///         self.quote_into(spec.schema().unwrap_or("dbo"), &mut table);
66///         table.push('.');
67///         self.quote_into(spec.table(), &mut table);
68///         let mut id = String::new();
69///         self.quote_into(spec.id().name(), &mut id);
70///         let Some(group) = spec.column(Role::Group) else {
71///             return Ok(Statement::new(
72///                 format!("DELETE FROM {table} WHERE {id} = @p1"),
73///                 [Param::Id],
74///             ));
75///         };
76///         let mut done = String::new();
77///         self.quote_into(group.name(), &mut done);
78///         Ok(Statement::new(
79///             format!("UPDATE {table} SET {done} = 'done' WHERE {id} = @p1"),
80///             [Param::Id],
81///         ))
82///     }
83///
84///     fn quote_into(&self, ident: &str, out: &mut String) {
85///         out.push('[');
86///         out.push_str(&ident.replace(']', "]]"));
87///         out.push(']');
88///     }
89/// #     fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { out.push_str("@p"); out.push_str(&index.to_string()); }
90/// #     fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedFetch { dialect: self.name() }) }
91/// #     fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
92/// #     fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
93/// #     fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
94/// #     fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
95/// #     fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
96/// #     fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { Err(StatementError::UnsupportedForm { dialect: self.name(), form: spec.form().name() }) }
97/// }
98/// ```
99#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
100pub struct TableSpec<'a> {
101    schema: Option<&'a str>,
102    table: &'a str,
103    id: Column<'a>,
104    grouping: Grouping<'a>,
105    partition_key: Option<Column<'a>>,
106    priority: Option<Column<'a>>,
107    retry_after: Option<Column<'a>>,
108    attempt: Option<Column<'a>>,
109    processed_at: Option<Column<'a>>,
110    headers: Option<Column<'a>>,
111    payload: Option<Column<'a>>,
112    data: &'a [Column<'a>],
113    fetched: &'a [Column<'a>],
114    form: Form<'a>,
115    opening: Opening,
116    select_all: bool,
117    database_clock: bool,
118}
ruststream_sqlx_dialect/spec.rs

ruststream_sqlx_dialect/
spec.rs
