ruststream_sqlx/inbox/named/by_name.rs
1//! By-name subscriptions as a feature of the dialect: the headers and times their statements read
2//! and bind, per database.
3
4#[cfg(all(
5 feature = "json",
6 any(feature = "postgres", feature = "mysql", feature = "sqlite")
7))]
8use std::collections::BTreeMap;
9
10use ruststream::HeaderMap;
11use ruststream_sqlx_dialect::Dialect;
12#[cfg(all(
13 any(feature = "postgres", feature = "mysql", feature = "sqlite"),
14 any(feature = "chrono", feature = "time")
15))]
16use sqlx::Arguments;
17use sqlx::error::BoxDynError;
18#[cfg(all(
19 feature = "json",
20 any(feature = "postgres", feature = "mysql", feature = "sqlite")
21))]
22use sqlx::types::Json;
23use sqlx::{Database, Error};
24#[cfg(all(
25 feature = "json",
26 any(feature = "postgres", feature = "mysql", feature = "sqlite")
27))]
28use sqlx::{Decode, Type, ValueRef};
29
30#[cfg(all(
31 feature = "json",
32 any(feature = "postgres", feature = "mysql", feature = "sqlite")
33))]
34use super::database::unread;
35use super::row::NamedTime;
36#[cfg(all(
37 feature = "json",
38 any(feature = "postgres", feature = "mysql", feature = "sqlite")
39))]
40use crate::HeaderColumn;
41#[cfg(any(
42 feature = "postgres",
43 feature = "mysql",
44 feature = "sqlite",
45 feature = "any"
46))]
47use crate::inbox::BuiltIn;
48
49/// Subscriptions by name on the database `DB`, as a feature of a dialect: the headers and the
50/// times their rows hold, which no struct of the service's own names.
51///
52/// A by-name subscription (`#[subscriber("emails")]`) reads the table the name's route leads to.
53/// Where the route's row leaves every event to the crate, the subscription reads the role
54/// columns into a row of the crate's own and binds its statements from there, with no box and no
55/// dynamic call per message. The ids, payloads, keys and attempts it reads are integers, text
56/// and bytes, which every sqlx driver decodes alike; the headers kept as JSON and the times are
57/// the database's own types, and this trait reads and binds them. A dialect that implements it
58/// for its database gets subscriptions by name; on a dialect that does not, a by-name mount does
59/// not compile, and its tables take [`InboxQueue`](crate::InboxQueue) descriptors.
60///
61/// [`BuiltIn<DB>`](crate::BuiltIn) implements it for each built-in database. A dialect of the
62/// service's own that wraps a built-in one delegates to it; one for a driver outside sqlx reads
63/// and binds through that driver.
64///
65/// # Examples
66///
67/// ```
68/// # #[cfg(feature = "postgres")]
69/// # mod demo {
70/// # use std::num::NonZeroUsize;
71/// # use ruststream::OutgoingMessage;
72/// # use ruststream_sqlx::dialect::{self, ClaimShape, Dialect, Opening, RowLock, Statement, StatementError, TableName, TableSpec};
73/// # use sqlx::PgConnection;
74/// use ruststream::HeaderMap;
75/// use ruststream_sqlx::prelude::*;
76/// use ruststream_sqlx::{BuiltIn, ByName, NamedTime};
77/// use serde::Deserialize;
78/// use sqlx::error::BoxDynError;
79/// use sqlx::postgres::{PgArguments, PgValueRef};
80/// use sqlx::{PgPool, Postgres};
81///
82/// /// A dialect of the service's own over Postgres.
83/// #[derive(Debug)]
84/// pub struct Audited;
85/// # impl Dialect for Audited {
86/// # fn name(&self) -> &'static str { "audited" }
87/// # fn quote_into(&self, ident: &str, out: &mut String) { dialect::Postgres.quote_into(ident, out) }
88/// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { dialect::Postgres.placeholder_into(index, out) }
89/// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.fetch(spec) }
90/// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.ack(spec) }
91/// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { dialect::Postgres.retry(spec) }
92/// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.retry_after(spec) }
93/// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.discard(spec) }
94/// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.dead_letter_group(spec) }
95/// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { dialect::Postgres.dead_letter_table(spec, target) }
96/// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.insert(spec) }
97/// # fn begin(&self, opening: Opening) -> Result<Option<&'static str>, StatementError> { dialect::Postgres.begin(opening) }
98/// # fn fifo_guard(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { dialect::Postgres.fifo_guard(spec) }
99/// # }
100/// # impl RowLock for Audited {
101/// # fn lock_claim(&self, spec: &TableSpec<'_>, shape: ClaimShape) -> Result<Statement, StatementError> { dialect::Postgres.lock_claim(spec, shape) }
102/// # }
103///
104/// // Its rows keep their headers and times as the built-in dialect reads them, so
105/// // `#[subscriber("emails")]` mounts on it.
106/// impl ByName<Postgres> for Audited {
107/// fn headers(value: PgValueRef<'_>) -> Result<HeaderMap, BoxDynError> {
108/// <BuiltIn<Postgres> as ByName<Postgres>>::headers(value)
109/// }
110///
111/// fn bind_time(arguments: &mut PgArguments, time: NamedTime) -> Result<(), sqlx::Error> {
112/// <BuiltIn<Postgres> as ByName<Postgres>>::bind_time(arguments, time)
113/// }
114/// }
115///
116/// #[derive(Inbox, sqlx::FromRow)]
117/// #[inbox(table = "email_jobs")]
118/// pub struct SendEmail {
119/// #[field(id, generated)]
120/// job_id: i64,
121/// #[field(group)]
122/// name: String,
123/// #[field(payload)]
124/// payload: Vec<u8>,
125/// }
126/// # impl Publish<Postgres> for SendEmail {
127/// # async fn publish(_: &mut PgConnection, _: &OutgoingMessage<'_>) -> Result<(), sqlx::Error> { Ok(()) }
128/// # }
129///
130/// #[derive(Deserialize)]
131/// struct Email {
132/// to: String,
133/// }
134///
135/// #[subscriber("emails")]
136/// async fn send(email: &Email) -> HandlerOutcome {
137/// tracing::info!(to = %email.to, "sending");
138/// HandlerOutcome::ack()
139/// }
140///
141/// pub fn app(pool: PgPool) -> RustStream {
142/// let broker = SqlxBroker::with_dialect(pool, Audited).route::<SendEmail>("emails");
143/// RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
144/// b.include(send);
145/// })
146/// }
147/// # }
148/// # fn main() {}
149/// ```
150#[diagnostic::on_unimplemented(
151 message = "the `{Self}` dialect has no subscriptions by name on `{DB}`",
152 label = "a subscription by name needs its dialect to read and bind its rows",
153 note = "implement `ByName<{DB}>` for the dialect, or subscribe through a descriptor: \
154 `#[subscriber(InboxQueue::<Row>::new(\"..\"))]`"
155)]
156pub trait ByName<DB: Database>: Dialect {
157 /// The headers a row keeps as a JSON object of strings.
158 ///
159 /// # Errors
160 ///
161 /// A column that holds no JSON, or the driver's decoding error.
162 fn headers(value: DB::ValueRef<'_>) -> Result<HeaderMap, BoxDynError>;
163
164 /// Binds `time`: the current time, a delayed retry's, or a lease's expiry.
165 ///
166 /// # Errors
167 ///
168 /// The driver's encoding error, or a time of a type the database does not bind.
169 fn bind_time(arguments: &mut DB::Arguments, time: NamedTime) -> Result<(), Error>;
170}
171
172/// Implements [`ByName`] for the built-in dialect of a database whose driver decodes JSON and
173/// binds the `chrono` and `time` types, with the features that bring them.
174#[cfg(any(feature = "postgres", feature = "mysql", feature = "sqlite"))]
175macro_rules! by_name {
176 ($database:ty) => {
177 impl ByName<$database> for BuiltIn<$database> {
178 fn headers(
179 value: <$database as Database>::ValueRef<'_>,
180 ) -> Result<HeaderMap, BoxDynError> {
181 #[cfg(feature = "json")]
182 {
183 type Headers = Json<BTreeMap<String, String>>;
184 let json = <Headers as Type<$database>>::compatible(&value.type_info());
185 if !json {
186 return Err(unread(&*value.type_info(), "JSON"));
187 }
188 let mut headers = <Headers as Decode<'_, $database>>::decode(value)?;
189 Ok(HeaderColumn::take_headers(&mut headers))
190 }
191 #[cfg(not(feature = "json"))]
192 {
193 let _ = value;
194 Err("headers kept as JSON need the `json` feature".into())
195 }
196 }
197
198 fn bind_time(
199 arguments: &mut <$database as Database>::Arguments,
200 time: NamedTime,
201 ) -> Result<(), Error> {
202 #[cfg(not(any(feature = "chrono", feature = "time")))]
203 let _ = arguments;
204 match time {
205 #[cfg(feature = "chrono")]
206 NamedTime::Chrono(at) => arguments.add(at).map_err(Error::Encode),
207 #[cfg(feature = "time")]
208 NamedTime::Time(at) => arguments.add(at).map_err(Error::Encode),
209 }
210 }
211 }
212 };
213}
214
215#[cfg(feature = "postgres")]
216by_name!(sqlx::Postgres);
217#[cfg(feature = "mysql")]
218by_name!(sqlx::MySql);
219#[cfg(feature = "sqlite")]
220by_name!(sqlx::Sqlite);
221
222// Why errors: `sqlx::Any` decodes no JSON and binds no time, so no row on an `AnyPool` has a
223// `headers` column of JSON or a time column, and a by-name subscription there reaches neither.
224#[cfg(feature = "any")]
225impl ByName<sqlx::Any> for BuiltIn<sqlx::Any> {
226 fn headers(value: <sqlx::Any as Database>::ValueRef<'_>) -> Result<HeaderMap, BoxDynError> {
227 let _ = value;
228 Err("an `AnyPool` decodes no JSON headers".into())
229 }
230
231 fn bind_time(
232 arguments: &mut <sqlx::Any as Database>::Arguments,
233 time: NamedTime,
234 ) -> Result<(), Error> {
235 let _ = (arguments, time);
236 Err(Error::Configuration("an `AnyPool` binds no time".into()))
237 }
238}