ruststream_sqlx/inbox/database/
built_in.rs

1//! The dialect built into the crate for each database: the broker's dialect unless the service
2//! passes its own.
3
4use std::fmt;
5use std::num::NonZeroUsize;
6
7#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
8use ruststream_sqlx_dialect as dialect;
9#[cfg(any(feature = "postgres", feature = "mysql", feature = "any"))]
10use ruststream_sqlx_dialect::RowLock;
11use ruststream_sqlx_dialect::{
12    Advisory, ClaimShape, Dialect, Lease, Opening, Statement, StatementError, TableName, TableSpec,
13};
14#[cfg(any(
15    feature = "postgres",
16    feature = "mysql",
17    feature = "sqlite",
18    feature = "any"
19))]
20use ruststream_sqlx_dialect::{Opens, level};
21#[cfg(feature = "any")]
22use sqlx::{Any, AnyConnection};
23#[cfg(feature = "mysql")]
24use sqlx::{MySql, MySqlConnection};
25#[cfg(feature = "postgres")]
26use sqlx::{PgConnection, Postgres};
27#[cfg(feature = "sqlite")]
28use sqlx::{Sqlite, SqliteConnection};
29
30use super::QueueDatabase;
31#[cfg(feature = "any")]
32use super::any::AnyDialect;
33
34/// The dialect built into the crate for the database `DB`: what [`SqlxBroker::new`] builds its
35/// statements with.
36///
37/// It builds them with the dialect of the crate's [`dialect`](crate::dialect) module that serves
38/// the database: [`Postgres`](crate::dialect::Postgres) on Postgres,
39/// [`MySql`](crate::dialect::MySql) on MySQL and MariaDB, [`Sqlite`](crate::dialect::Sqlite) on
40/// SQLite. On an `AnyPool` it is the dialect of the database the pool reaches, among those whose
41/// features are on, picked when the broker connects.
42///
43/// It serves the forms its database serves, through the traits of the dialect module: every
44/// `BuiltIn` implements [`Lease`] and [`Advisory`], and `BuiltIn<Postgres>`, `BuiltIn<MySql>` and
45/// `BuiltIn<Any>` implement [`RowLock`]. A SQLite table in the row lock form therefore does not
46/// compile. An `AnyPool` names its database only when the broker connects, so on a SQLite backend
47/// the subscription to a row lock table stops when it starts, with
48/// [`StatementError::UnsupportedForm`].
49///
50/// It opens transactions at the isolation levels and SQLite modes its database keeps, one
51/// [`Opens`](crate::dialect::Opens) per level: `BuiltIn<Postgres>` READ COMMITTED, REPEATABLE
52/// READ and SERIALIZABLE, `BuiltIn<MySql>` those and READ UNCOMMITTED, `BuiltIn<Sqlite>` the three
53/// modes. A table that names another does not compile. `BuiltIn<Any>` takes all seven, and the
54/// picked backend's dialect refuses the ones its database lacks when the subscription starts,
55/// with [`StatementError::UnsupportedOpening`].
56///
57/// [`SqlxBroker::new`]: crate::SqlxBroker::new
58///
59/// # Examples
60///
61/// ```
62/// # #[cfg(feature = "postgres")]
63/// # mod demo {
64/// use ruststream_sqlx::BuiltIn;
65/// use ruststream_sqlx::prelude::*;
66/// use serde::Deserialize;
67/// use sqlx::{PgPool, Postgres};
68///
69/// #[derive(Inbox, sqlx::FromRow)]
70/// #[inbox(table = "email_jobs")]
71/// pub struct SendEmail {
72///     #[field(id, generated)]
73///     job_id: i64,
74///     #[field(group)]
75///     name: String,
76///     #[field(payload)]
77///     payload: Vec<u8>,
78/// }
79///
80/// #[derive(Deserialize)]
81/// struct Email {
82///     to: String,
83/// }
84///
85/// #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
86/// async fn send(email: &Email) -> HandlerOutcome {
87///     tracing::info!(to = %email.to, "sending");
88///     HandlerOutcome::ack()
89/// }
90///
91/// // `SqlxBroker<Postgres>` names the same type: the built-in dialect is the default.
92/// pub fn broker(pool: PgPool) -> SqlxBroker<Postgres, BuiltIn<Postgres>> {
93///     SqlxBroker::new(pool)
94/// }
95///
96/// pub fn app(pool: PgPool) -> RustStream {
97///     RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker(pool), |b| {
98///         b.include(send);
99///     })
100/// }
101/// # }
102/// # fn main() {}
103/// ```
104pub struct BuiltIn<DB: BuiltInDialect>(DB::Picked);
105
106impl<DB: BuiltInDialect> BuiltIn<DB> {
107    /// The built-in dialect that `picked` builds the statements of.
108    #[cfg(any(
109        feature = "postgres",
110        feature = "mysql",
111        feature = "sqlite",
112        feature = "any"
113    ))]
114    pub(crate) const fn new(picked: DB::Picked) -> Self {
115        Self(picked)
116    }
117}
118
119impl<DB: BuiltInDialect> Clone for BuiltIn<DB> {
120    fn clone(&self) -> Self {
121        *self
122    }
123}
124
125impl<DB: BuiltInDialect> Copy for BuiltIn<DB> {}
126
127impl<DB: BuiltInDialect> fmt::Debug for BuiltIn<DB> {
128    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
129        f.debug_tuple("BuiltIn").field(&self.0.name()).finish()
130    }
131}
132
133impl<DB: BuiltInDialect> Dialect for BuiltIn<DB> {
134    fn name(&self) -> &'static str {
135        self.0.name()
136    }
137
138    fn quote_into(&self, ident: &str, out: &mut String) {
139        self.0.quote_into(ident, out);
140    }
141
142    fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) {
143        self.0.placeholder_into(index, out);
144    }
145
146    fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
147        self.0.fetch(spec)
148    }
149
150    fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
151        self.0.ack(spec)
152    }
153
154    fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> {
155        self.0.retry(spec)
156    }
157
158    fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
159        self.0.retry_after(spec)
160    }
161
162    fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
163        self.0.discard(spec)
164    }
165
166    fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
167        self.0.dead_letter_group(spec)
168    }
169
170    fn dead_letter_table(
171        &self,
172        spec: &TableSpec<'_>,
173        target: TableName<'_>,
174    ) -> Result<Vec<Statement>, StatementError> {
175        self.0.dead_letter_table(spec, target)
176    }
177
178    fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
179        self.0.insert(spec)
180    }
181
182    fn server_version(&self) -> Option<&'static str> {
183        self.0.server_version()
184    }
185
186    fn check_server(&self, spec: &TableSpec<'_>, version: &str) -> Result<(), StatementError> {
187        self.0.check_server(spec, version)
188    }
189
190    fn begin(&self, opening: Opening) -> Result<Option<&'static str>, StatementError> {
191        self.0.begin(opening)
192    }
193
194    fn savepoint(&self) -> &'static str {
195        self.0.savepoint()
196    }
197
198    fn rollback_to_savepoint(&self) -> &'static str {
199        self.0.rollback_to_savepoint()
200    }
201
202    fn fifo_guard(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> {
203        self.0.fifo_guard(spec)
204    }
205}
206
207impl<DB: BuiltInDialect> Lease for BuiltIn<DB> {
208    fn lease_claim(
209        &self,
210        spec: &TableSpec<'_>,
211        shape: ClaimShape,
212    ) -> Result<Statement, StatementError> {
213        self.0.lease_claim(spec, shape)
214    }
215
216    fn extend(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
217        self.0.extend(spec)
218    }
219
220    fn stamp(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
221        self.0.stamp(spec)
222    }
223
224    fn claim_writes_lease(&self) -> bool {
225        self.0.claim_writes_lease()
226    }
227
228    fn claim_counts_attempt(&self, spec: &TableSpec<'_>) -> bool {
229        self.0.claim_counts_attempt(spec)
230    }
231
232    fn begin_lease_claim(&self) -> Option<&'static str> {
233        self.0.begin_lease_claim()
234    }
235}
236
237impl<DB: BuiltInDialect> Advisory for BuiltIn<DB> {
238    fn advisory_claim(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
239        self.0.advisory_claim(spec)
240    }
241
242    fn lock(&self) -> Option<Statement> {
243        self.0.lock()
244    }
245
246    fn unlock(&self) -> Option<Statement> {
247        self.0.unlock()
248    }
249
250    fn take(
251        &self,
252        spec: &TableSpec<'_>,
253        shape: ClaimShape,
254    ) -> Result<Vec<Statement>, StatementError> {
255        self.0.take(spec, shape)
256    }
257}
258
259#[cfg(feature = "postgres")]
260impl RowLock for BuiltIn<Postgres> {
261    fn lock_claim(
262        &self,
263        spec: &TableSpec<'_>,
264        shape: ClaimShape,
265    ) -> Result<Statement, StatementError> {
266        self.0.lock_claim(spec, shape)
267    }
268}
269
270#[cfg(feature = "mysql")]
271impl RowLock for BuiltIn<MySql> {
272    fn lock_claim(
273        &self,
274        spec: &TableSpec<'_>,
275        shape: ClaimShape,
276    ) -> Result<Statement, StatementError> {
277        self.0.lock_claim(spec, shape)
278    }
279}
280
281#[cfg(feature = "any")]
282impl RowLock for BuiltIn<Any> {
283    fn lock_claim(
284        &self,
285        spec: &TableSpec<'_>,
286        shape: ClaimShape,
287    ) -> Result<Statement, StatementError> {
288        self.0.lock_claim(spec, shape)
289    }
290}
291
292#[cfg(feature = "postgres")]
293impl Opens<level::ReadCommitted> for BuiltIn<Postgres> {}
294
295#[cfg(feature = "postgres")]
296impl Opens<level::RepeatableRead> for BuiltIn<Postgres> {}
297
298#[cfg(feature = "postgres")]
299impl Opens<level::Serializable> for BuiltIn<Postgres> {}
300
301#[cfg(feature = "mysql")]
302impl Opens<level::ReadUncommitted> for BuiltIn<MySql> {}
303
304#[cfg(feature = "mysql")]
305impl Opens<level::ReadCommitted> for BuiltIn<MySql> {}
306
307#[cfg(feature = "mysql")]
308impl Opens<level::RepeatableRead> for BuiltIn<MySql> {}
309
310#[cfg(feature = "mysql")]
311impl Opens<level::Serializable> for BuiltIn<MySql> {}
312
313#[cfg(feature = "sqlite")]
314impl Opens<level::Deferred> for BuiltIn<Sqlite> {}
315
316#[cfg(feature = "sqlite")]
317impl Opens<level::Immediate> for BuiltIn<Sqlite> {}
318
319#[cfg(feature = "sqlite")]
320impl Opens<level::Exclusive> for BuiltIn<Sqlite> {}
321
322// Why every level and mode compiles on `Any`: an `AnyPool` names its database only when the
323// broker connects, so the type cannot say which of them the database opens. The picked backend's
324// dialect refuses the ones it lacks when the subscription to such a table starts.
325#[cfg(feature = "any")]
326impl Opens<level::ReadUncommitted> for BuiltIn<Any> {}
327
328#[cfg(feature = "any")]
329impl Opens<level::ReadCommitted> for BuiltIn<Any> {}
330
331#[cfg(feature = "any")]
332impl Opens<level::RepeatableRead> for BuiltIn<Any> {}
333
334#[cfg(feature = "any")]
335impl Opens<level::Serializable> for BuiltIn<Any> {}
336
337#[cfg(feature = "any")]
338impl Opens<level::Deferred> for BuiltIn<Any> {}
339
340#[cfg(feature = "any")]
341impl Opens<level::Immediate> for BuiltIn<Any> {}
342
343#[cfg(feature = "any")]
344impl Opens<level::Exclusive> for BuiltIn<Any> {}
345
346/// A database whose dialect is built into this crate, so `SqlxBroker::new` needs no dialect of
347/// the service's own.
348///
349/// Postgres (feature `postgres`), MySQL with MariaDB (feature `mysql`) and SQLite (feature
350/// `sqlite`) have one, [`BuiltIn<DB>`](BuiltIn). `Any` (feature `any`) takes the dialect of the
351/// database its pool reaches, among those whose features are on, and the broker picks it when it
352/// connects. A row on an `AnyPool` holds only the types `sqlx::Any` carries, and no time is among
353/// them, so a table with `locked_until`, `retry_after` or `processed_at` is out of its reach.
354///
355/// # Examples
356///
357/// ```
358/// # #[cfg(all(feature = "any", feature = "postgres", feature = "mysql"))]
359/// # mod demo {
360/// use ruststream_sqlx::prelude::*;
361/// use serde::Deserialize;
362/// use sqlx::AnyPool;
363///
364/// // The columns `sqlx::Any` carries: integers, text and bytes, no time.
365/// #[derive(Inbox, sqlx::FromRow)]
366/// #[inbox(table = "email_jobs")]
367/// pub struct SendEmail {
368///     #[field(id, generated)]
369///     job_id: i64,
370///     #[field(group)]
371///     name: String,
372///     #[field(attempt, generated)]
373///     attempt: i32,
374///     #[field(payload)]
375///     payload: Vec<u8>,
376/// }
377///
378/// #[derive(Deserialize)]
379/// struct Email {
380///     to: String,
381/// }
382///
383/// #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
384/// async fn send(email: &Email) -> HandlerOutcome {
385///     tracing::info!(to = %email.to, "sending");
386///     HandlerOutcome::ack()
387/// }
388///
389/// // One build for every deployment: `Any` is a `BuiltInDialect`, and the broker takes the
390/// // dialect of the database the URL names, Postgres or MySQL, when it connects. `main` installs
391/// // sqlx's drivers (`sqlx::any::install_default_drivers`) before it builds the pool.
392/// pub fn app(pool: AnyPool) -> RustStream {
393///     RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
394///         b.include(send);
395///     })
396/// }
397/// # }
398/// # fn main() {}
399/// ```
400#[diagnostic::on_unimplemented(
401    message = "`{Self}` has no dialect built into the crate",
402    label = "no built-in dialect for this database",
403    note = "build the broker with a dialect of the service's own: \
404            `SqlxBroker::with_dialect(pool, dialect)`, a `SqlxBroker<{Self}, YourDialect>`"
405)]
406pub trait BuiltInDialect: QueueDatabase {
407    /// The dialect of the crate's `dialect` module that builds the database's statements.
408    /// Machinery; [`BuiltIn`] holds it.
409    #[doc(hidden)]
410    type Picked: Lease + Advisory + Copy + 'static;
411
412    /// The dialect that builds the statements of the database `conn` reaches, or `None` when no
413    /// built-in dialect serves it: an `AnyPool`'s backend whose feature is off.
414    ///
415    /// Postgres, MySQL and SQLite answer without reading `conn`; the broker asks with the
416    /// connection it checks when it connects.
417    fn dialect(conn: &Self::Connection) -> Option<BuiltIn<Self>>;
418
419    /// The name of the database `conn` reaches, which an error names when no built-in dialect
420    /// serves it. Machinery; the broker calls it.
421    #[doc(hidden)]
422    fn backend(conn: &Self::Connection) -> &str {
423        let _ = conn;
424        Self::NAME
425    }
426}
427
428#[cfg(feature = "postgres")]
429impl BuiltInDialect for Postgres {
430    type Picked = dialect::Postgres;
431
432    fn dialect(_: &PgConnection) -> Option<BuiltIn<Self>> {
433        Some(BuiltIn::new(dialect::Postgres))
434    }
435}
436
437#[cfg(feature = "mysql")]
438impl BuiltInDialect for MySql {
439    type Picked = dialect::MySql;
440
441    fn dialect(_: &MySqlConnection) -> Option<BuiltIn<Self>> {
442        Some(BuiltIn::new(dialect::MySql))
443    }
444}
445
446#[cfg(feature = "sqlite")]
447impl BuiltInDialect for Sqlite {
448    type Picked = dialect::Sqlite;
449
450    fn dialect(_: &SqliteConnection) -> Option<BuiltIn<Self>> {
451        Some(BuiltIn::new(dialect::Sqlite))
452    }
453}
454
455#[cfg(feature = "any")]
456impl BuiltInDialect for Any {
457    type Picked = AnyDialect;
458
459    fn dialect(conn: &AnyConnection) -> Option<BuiltIn<Self>> {
460        AnyDialect::of(conn.backend_name()).map(BuiltIn::new)
461    }
462
463    fn backend(conn: &AnyConnection) -> &str {
464        conn.backend_name()
465    }
466}
467
468/// The name an `AnyConnection` reports for a Postgres backend: its sqlx driver's
469/// `Database::NAME`.
470#[cfg(feature = "any")]
471pub(crate) const POSTGRES_BACKEND: &str = "PostgreSQL";
472
473/// The name an `AnyConnection` reports for a MySQL or MariaDB backend.
474#[cfg(feature = "any")]
475pub(crate) const MYSQL_BACKEND: &str = "MySQL";
476
477/// The name an `AnyConnection` reports for a SQLite backend.
478#[cfg(feature = "any")]
479pub(crate) const SQLITE_BACKEND: &str = "SQLite";
480
481#[cfg(all(test, feature = "any"))]
482mod tests {
483    use sqlx::Database;
484
485    use super::{MYSQL_BACKEND, POSTGRES_BACKEND, SQLITE_BACKEND};
486
487    #[test]
488    fn the_backend_names_are_those_sqlx_reports() {
489        assert_eq!(POSTGRES_BACKEND, <sqlx::Postgres as Database>::NAME);
490        assert_eq!(MYSQL_BACKEND, <sqlx::MySql as Database>::NAME);
491        assert_eq!(SQLITE_BACKEND, <sqlx::Sqlite as Database>::NAME);
492    }
493}
ruststream_sqlx/inbox/database/built_in.rs

ruststream_sqlx/inbox/database/
built_in.rs
