ruststream_sqlx/inbox/events.rs
1//! One trait per event of the queue: what a service implements when its table names the event as
2//! its own, in `#[inbox(custom(..))]` or with `InboxSpec::own`, and `Publish`, which has no
3//! default.
4
5use std::future::Future;
6use std::time::Duration;
7
8use ruststream::OutgoingMessage;
9use sqlx::{Database, Error};
10
11use super::InboxRow;
12use super::time::LeaseRow;
13
14/// Claims up to `limit` rows of the queue `queue` and returns their ids, inside the claim's
15/// transaction.
16///
17/// The dialect builds it from the table's description; a table names `claim` as its own to take it
18/// over (`custom(claim)`, or [`own::Claim`](crate::spec::own::Claim) by hand), for a database
19/// without a built-in dialect or a claim of its own. The crate then reads the
20/// rows with the service's [`Fetch`], or with its own fetch by a list of ids, which Postgres
21/// alone runs: on MySQL and SQLite a claim of the service's own comes with a `Fetch` of its own,
22/// and a subscription without one stops at startup. In the lease form the crate also leases each
23/// id the claim returns, in the same transaction, and passes over an id whose row another lease
24/// holds.
25///
26/// # Examples
27///
28/// ```
29/// # #[cfg(feature = "postgres")]
30/// # mod demo {
31/// use ruststream_sqlx::{Claim, Inbox};
32/// use sqlx::{PgConnection, Postgres};
33///
34/// #[derive(Inbox, sqlx::FromRow)]
35/// #[inbox(table = "jobs", custom(claim))]
36/// pub struct Job {
37/// #[field(id)]
38/// id: i64,
39/// #[field(payload)]
40/// payload: Vec<u8>,
41/// }
42///
43/// impl Claim<Postgres> for Job {
44/// async fn claim(
45/// conn: &mut PgConnection,
46/// _queue: &str,
47/// limit: i64,
48/// ) -> Result<Vec<i64>, sqlx::Error> {
49/// // The oldest rows first, skipping the ones another worker holds.
50/// sqlx::query_scalar("SELECT id FROM jobs ORDER BY id LIMIT $1 FOR UPDATE SKIP LOCKED")
51/// .bind(limit)
52/// .fetch_all(conn)
53/// .await
54/// }
55/// }
56/// # }
57/// # fn main() {}
58/// ```
59#[diagnostic::on_unimplemented(
60 message = "`{Self}` runs its own `claim` (`custom(claim)` in `#[inbox(..)]`, or `own::Claim` in its `InboxSpec`) and does not implement `Claim<{DB}>`",
61 label = "the service's own claim is missing",
62 note = "implement `Claim<{DB}>` for `{Self}`, or drop `claim` from `custom(..)` (by hand: `.own::<own::Claim>()` and `own::Claim` in `type Table`)"
63)]
64pub trait Claim<DB: Database>: InboxRow {
65 /// Claims up to `limit` rows and returns their ids.
66 ///
67 /// # Errors
68 ///
69 /// The database's error; the subscription waits one second and claims again.
70 fn claim(
71 conn: &mut DB::Connection,
72 queue: &str,
73 limit: i64,
74 ) -> impl Future<Output = Result<Vec<Self::Id>, Error>> + Send;
75}
76
77/// Reads the rows of claimed ids, on the claim's connection.
78///
79/// The dialect builds it, for a flat table and for a message assembled from header fields; a table
80/// names `fetch` as its own (`custom(fetch)`, or [`own::Fetch`](crate::spec::own::Fetch) by hand)
81/// to assemble messages itself, from other tables
82/// ([a fetch over a join](crate#a-fetch-over-a-join)). Rows are matched to the claimed ids by their `id` field; a
83/// claimed id with no row settles by the decode-failure policy before its handler runs, and the
84/// log names the id. It runs inside the claim's transaction, or right after a lease claim that
85/// leased the rows and committed in one statement, as on Postgres and SQLite. In the advisory lock
86/// form it runs once per row, with that row's id alone, after the take that counted its attempt.
87/// Either way a delivery reports the attempt its row held before the claim. It runs only after a
88/// claim that took ids, so `ids` is never empty.
89///
90/// # Examples
91///
92/// ```
93/// # #[cfg(feature = "postgres")]
94/// # mod demo {
95/// use ruststream_sqlx::{Fetch, Inbox};
96/// use sqlx::{PgConnection, Postgres};
97///
98/// /// A job whose message is the body of the order it refers to.
99/// #[derive(Inbox, sqlx::FromRow)]
100/// #[inbox(table = "jobs", custom(fetch))]
101/// pub struct Job {
102/// #[field(id)]
103/// id: i64,
104/// #[field(payload)]
105/// payload: Vec<u8>,
106/// }
107///
108/// impl Fetch<Postgres> for Job {
109/// async fn fetch(conn: &mut PgConnection, ids: &[i64]) -> Result<Vec<Self>, sqlx::Error> {
110/// sqlx::query_as(
111/// "SELECT j.id, o.body AS payload FROM jobs j JOIN orders o ON o.id = j.order_id \
112/// WHERE j.id = ANY($1)",
113/// )
114/// .bind(ids)
115/// .fetch_all(conn)
116/// .await
117/// }
118/// }
119/// # }
120/// # fn main() {}
121/// ```
122#[diagnostic::on_unimplemented(
123 message = "`{Self}` runs its own `fetch` (`custom(fetch)` in `#[inbox(..)]`, or `own::Fetch` in its `InboxSpec`) and does not implement `Fetch<{DB}>`",
124 label = "the service's own fetch is missing",
125 note = "implement `Fetch<{DB}>` for `{Self}`, or drop `fetch` from `custom(..)` (by hand: `.own::<own::Fetch>()` and `own::Fetch` in `type Table`)"
126)]
127pub trait Fetch<DB: Database>: InboxRow {
128 /// The rows of `ids`, in any order.
129 ///
130 /// # Errors
131 ///
132 /// The database's error; the claim fails and its rows return to the queue.
133 fn fetch(
134 conn: &mut DB::Connection,
135 ids: &[Self::Id],
136 ) -> impl Future<Output = Result<Vec<Self>, Error>> + Send;
137}
138
139macro_rules! settle_event {
140 (
141 $(#[$doc:meta])*
142 $trait:ident, $method:ident, $event:literal,
143 gate = $gate:literal, $(uses = $uses:literal,)? $(fields = $fields:literal,)? body = $body:literal,
144 message = $message:literal, note = $note:literal
145 $(, $extra:ident: $extra_ty:ty = $extra_value:literal)?
146 ) => {
147 $(#[$doc])*
148 #[doc = ""]
149 #[doc = "# Examples"]
150 #[doc = ""]
151 #[doc = "```"]
152 #[doc = concat!("# #[cfg(", $gate, ")]")]
153 #[doc = "# mod demo {"]
154 #[doc = "use std::time::Duration;"]
155 #[doc = ""]
156 #[doc = concat!("use ruststream_sqlx::{", stringify!($trait), ", Inbox};")]
157 #[doc = "use sqlx::{PgConnection, Postgres};"]
158 $(#[doc = $uses])?
159 #[doc = ""]
160 #[doc = "#[derive(Inbox, sqlx::FromRow)]"]
161 #[doc = concat!("#[inbox(table = \"jobs\", custom(", $event, "))]")]
162 #[doc = "pub struct Job {"]
163 #[doc = " #[field(id)]"]
164 #[doc = " id: i64,"]
165 $(#[doc = $fields])?
166 #[doc = " #[field(payload)]"]
167 #[doc = " payload: Vec<u8>,"]
168 #[doc = "}"]
169 #[doc = ""]
170 #[doc = concat!("impl ", stringify!($trait), "<Postgres> for Job {")]
171 #[doc = concat!(" async fn ", stringify!($method), "(")]
172 #[doc = " conn: &mut PgConnection,"]
173 #[doc = " id: &i64,"]
174 $(#[doc = concat!(" ", stringify!($extra), ": ", stringify!($extra_ty), ",")])?
175 #[doc = " ) -> Result<(), sqlx::Error> {"]
176 #[doc = $body]
177 #[doc = " Ok(())"]
178 #[doc = " }"]
179 #[doc = "}"]
180 #[doc = "# }"]
181 #[doc = "# fn main() {}"]
182 #[doc = "```"]
183 #[diagnostic::on_unimplemented(
184 message = $message,
185 label = "the service's own event is missing",
186 note = $note
187 )]
188 pub trait $trait<DB: Database>: InboxRow {
189 #[doc = concat!("Runs the `", $event, "` event for the row `id` inside a transaction the crate commits: the claim's, or in the lease form one that first confirms the row still holds the delivery's lease.")]
190 #[doc = ""]
191 #[doc = "# Errors"]
192 #[doc = ""]
193 #[doc = "The database's error; the transaction rolls back, and the row returns to the queue at once, or in the lease form once its lease runs out."]
194 fn $method(
195 conn: &mut DB::Connection,
196 id: &Self::Id,
197 $($extra: $extra_ty,)?
198 ) -> impl Future<Output = Result<(), Error>> + Send;
199 }
200 };
201}
202
203settle_event!(
204 /// Acknowledges a row: by default it deletes the row, or sets `processed_at`.
205 ///
206 /// The service's own acknowledgement takes the row out of what the claim selects: it deletes
207 /// the row, moves it, or marks it in a column the claim passes over. A row left claimable is
208 /// delivered again.
209 Ack, ack, "ack",
210 gate = "feature = \"postgres\"",
211 body = " // The finished job moves to an archive in one statement.
212 sqlx::query(
213 \"WITH done AS (DELETE FROM jobs WHERE id = $1 RETURNING *) \\
214 INSERT INTO jobs_done SELECT * FROM done\",
215 )
216 .bind(id)
217 .execute(conn)
218 .await?;",
219 message = "`{Self}` runs its own `ack` (`custom(ack)` in `#[inbox(..)]`, or `own::Ack` in its `InboxSpec`) and does not implement `Ack<{DB}>`",
220 note = "implement `Ack<{DB}>` for `{Self}`, or drop `ack` from `custom(..)` (by hand: `.own::<own::Ack>()` and `own::Ack` in `type Table`)"
221);
222
223settle_event!(
224 /// Releases a row for another attempt at once: by default it counts the attempt, or leaves the
225 /// row for the rollback to release; in the lease form it clears the lease.
226 ///
227 /// In the lease form the row comes back once `locked_until` no longer holds it: the service's
228 /// own retry clears the column to release the row at once, and otherwise the row waits for its
229 /// lease to run out.
230 Retry, retry, "retry",
231 gate = "feature = \"postgres\"",
232 body = " sqlx::query(\"UPDATE jobs SET tries = tries + 1 WHERE id = $1\")
233 .bind(id)
234 .execute(conn)
235 .await?;",
236 message = "`{Self}` runs its own `retry` (`custom(retry)` in `#[inbox(..)]`, or `own::Retry` in its `InboxSpec`) and does not implement `Retry<{DB}>`",
237 note = "implement `Retry<{DB}>` for `{Self}`, or drop `retry` from `custom(..)` (by hand: `.own::<own::Retry>()` and `own::Retry` in `type Table`)"
238);
239
240settle_event!(
241 /// Releases a row for another attempt after `delay`: by default it sets `retry_after`.
242 ///
243 /// A row with this event, or a `retry_after` field, is redelivered by the database's own
244 /// clock. The service's own event delays a row only where the claim passes over it until
245 /// then: the default claim reads the `retry_after` field.
246 RetryAfter, retry_after, "retry_after",
247 gate = "all(feature = \"postgres\", feature = \"chrono\")",
248 uses = "use chrono::{DateTime, Utc};",
249 fields = " #[field(retry_after)]
250 retry_after: DateTime<Utc>,",
251 body = " // The delay counts from the database's clock, whatever the host's says.
252 sqlx::query(
253 \"UPDATE jobs SET retry_after = now() + make_interval(secs => $2) WHERE id = $1\",
254 )
255 .bind(id)
256 .bind(delay.as_secs_f64())
257 .execute(conn)
258 .await?;",
259 message = "`{Self}` runs its own `retry_after` (`custom(retry_after)` in `#[inbox(..)]`, or `own::RetryAfter` in its `InboxSpec`) and does not implement `RetryAfter<{DB}>`",
260 note = "implement `RetryAfter<{DB}>` for `{Self}`, or drop `retry_after` from `custom(..)` (by hand: `.own::<own::RetryAfter>()` and `own::RetryAfter` in `type Table`)",
261 delay: Duration = "Duration::from_secs(30)"
262);
263
264settle_event!(
265 /// Drops a row: by default it deletes the row, or sets `processed_at`.
266 ///
267 /// The service's own drop takes the row out of what the claim selects, as an acknowledgement
268 /// does.
269 Discard, discard, "discard",
270 gate = "feature = \"postgres\"",
271 body = " sqlx::query(\"DELETE FROM jobs WHERE id = $1\")
272 .bind(id)
273 .execute(conn)
274 .await?;",
275 message = "`{Self}` runs its own `discard` (`custom(discard)` in `#[inbox(..)]`, or `own::Discard` in its `InboxSpec`) and does not implement `Discard<{DB}>`",
276 note = "implement `Discard<{DB}>` for `{Self}`, or drop `discard` from `custom(..)` (by hand: `.own::<own::Discard>()` and `own::Discard` in `type Table`)"
277);
278
279settle_event!(
280 /// Moves a row whose attempts are spent to `destination`: by default it moves the row to that
281 /// group (on a table with a `group` column) or into that table.
282 ///
283 /// The service's own move takes the row out of what the claim selects; a row copied and left
284 /// behind is moved again at every delivery.
285 DeadLetter, dead_letter, "dead_letter",
286 gate = "feature = \"postgres\"",
287 body = " // One dead-letter table serves the whole service, whatever the destination names.
288 sqlx::query(
289 \"WITH dead AS (DELETE FROM jobs WHERE id = $1 RETURNING *) \\
290 INSERT INTO dead_jobs SELECT * FROM dead\",
291 )
292 .bind(id)
293 .execute(conn)
294 .await?;
295 tracing::warn!(id, destination, \"a job's attempts are spent\");",
296 message = "`{Self}` runs its own `dead_letter` (`custom(dead_letter)` in `#[inbox(..)]`, or `own::DeadLetter` in its `InboxSpec`) and does not implement `DeadLetter<{DB}>`",
297 note = "implement `DeadLetter<{DB}>` for `{Self}`, or drop `dead_letter` from `custom(..)` (by hand: `.own::<own::DeadLetter>()` and `own::DeadLetter` in `type Table`)",
298 destination: &str = "\"jobs.dead\""
299);
300
301/// Extends the lease on a row: writes `until` into `locked_until` while the row still holds
302/// `held`, and says whether it did.
303///
304/// The dialect builds it for every table in the lease form; a table names `extend` as its own
305/// (`custom(extend)`, or [`own::Extend`](crate::spec::own::Extend) by hand) to take it over, for a
306/// database without a built-in dialect. A
307/// subscription runs it each half lease for every delivery in work, on one connection. The crate
308/// also runs it with `until` equal to `held` to confirm a delivery's lease, inside the transaction
309/// where an event of the service's own then runs.
310///
311/// # Examples
312///
313/// ```
314/// # #[cfg(all(feature = "postgres", feature = "chrono"))]
315/// # mod demo {
316/// use chrono::{DateTime, Utc};
317/// use ruststream_sqlx::{Extend, Inbox};
318/// use sqlx::{PgConnection, Postgres};
319///
320/// #[derive(Inbox, sqlx::FromRow)]
321/// #[inbox(table = "jobs", custom(extend))]
322/// pub struct Job {
323/// #[field(id)]
324/// id: i64,
325/// #[field(locked_until)]
326/// locked_until: Option<DateTime<Utc>>,
327/// #[field(payload)]
328/// payload: Vec<u8>,
329/// }
330///
331/// impl Extend<Postgres> for Job {
332/// async fn extend(
333/// conn: &mut PgConnection,
334/// id: &i64,
335/// held: &DateTime<Utc>,
336/// until: &DateTime<Utc>,
337/// ) -> Result<bool, sqlx::Error> {
338/// // The row keeps its lease only while it holds the expiry the delivery knows.
339/// let extended =
340/// sqlx::query("UPDATE jobs SET locked_until = $1 WHERE id = $2 AND locked_until = $3")
341/// .bind(until)
342/// .bind(id)
343/// .bind(held)
344/// .execute(conn)
345/// .await?;
346/// Ok(extended.rows_affected() == 1)
347/// }
348/// }
349/// # }
350/// # fn main() {}
351/// ```
352#[diagnostic::on_unimplemented(
353 message = "`{Self}` runs its own `extend` (`custom(extend)` in `#[inbox(..)]`, or `own::Extend` in its `InboxSpec`) and does not implement `Extend<{DB}>`",
354 label = "the service's own extension is missing",
355 note = "implement `Extend<{DB}>` for `{Self}`, or drop `extend` from `custom(..)` (by hand: `.own::<own::Extend>()` and `own::Extend` in `type Table`)"
356)]
357pub trait Extend<DB: Database>: LeaseRow {
358 /// Writes `until` into the lease of the row `id` while it still holds `held`; `true` when it
359 /// did, `false` when the row no longer holds `held`.
360 ///
361 /// # Errors
362 ///
363 /// The database's error.
364 fn extend(
365 conn: &mut DB::Connection,
366 id: &Self::Id,
367 held: &Self::Lease,
368 until: &Self::Lease,
369 ) -> impl Future<Output = Result<bool, Error>> + Send;
370}
371
372/// Takes the advisory lock on a row's key for the session that will hold its delivery, without
373/// waiting: the `lock` event of the advisory lock form.
374///
375/// The dialect takes the lock itself on Postgres and MySQL, and the process keeps the keys in work
376/// on SQLite. A table names `lock` and `unlock` as its own together (`custom(lock, unlock)`, or
377/// [`own::Lock`](crate::spec::own::Lock) and [`own::Unlock`](crate::spec::own::Unlock) by hand) to
378/// run the service's own SQL for them instead, as a database without a built-in dialect does, MSSQL with
379/// `sp_getapplock` and `sp_releaseapplock`; the dialect still selects the candidates and takes
380/// each row. A claim calls it once per candidate with the key the template renders, on the
381/// connection that holds the delivery; [`Unlock`] releases the lock when the delivery settles,
382/// and where the delivery is dropped unsettled the connection closes after that unlock. With a
383/// lock of the service's own the process keeps no registry of keys, on SQLite too.
384///
385/// The dialect's candidate select cannot see a lock of the service's own, so a claim reads as many
386/// candidates past its limit as the subscription has deliveries in work: a key its deliveries hold
387/// does not hold back the rows behind it. A key the service's lock holds elsewhere, in another
388/// process or system, is passed over within that margin only: while such keys head the claim
389/// order, the claim may end empty and wait for its next poll.
390///
391/// # Examples
392///
393/// ```no_run
394/// # #[cfg(feature = "postgres")]
395/// # mod demo {
396/// use ruststream_sqlx::{Inbox, Lock, Unlock};
397/// use sqlx::{PgConnection, Postgres};
398///
399/// #[derive(Inbox, sqlx::FromRow)]
400/// #[inbox(table = "jobs", advisory_lock = "jobs-{id}", custom(lock, unlock))]
401/// pub struct Job {
402/// #[field(id)]
403/// id: i64,
404/// #[field(payload)]
405/// payload: Vec<u8>,
406/// }
407///
408/// // The pair MSSQL writes with `sp_getapplock` and `sp_releaseapplock`, written here with
409/// // Postgres's session locks on a 32-bit hash of the key.
410/// impl Lock<Postgres> for Job {
411/// async fn lock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
412/// sqlx::query_scalar("SELECT pg_try_advisory_lock(hashtext($1))")
413/// .bind(key)
414/// .fetch_one(conn)
415/// .await
416/// }
417/// }
418///
419/// impl Unlock<Postgres> for Job {
420/// async fn unlock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
421/// sqlx::query_scalar("SELECT pg_advisory_unlock(hashtext($1))")
422/// .bind(key)
423/// .fetch_one(conn)
424/// .await
425/// }
426/// }
427/// # }
428/// # fn main() {}
429/// ```
430#[diagnostic::on_unimplemented(
431 message = "`{Self}` runs its own `lock` (`custom(lock)` in `#[inbox(..)]`, or `own::Lock` in its `InboxSpec`) and does not implement `Lock<{DB}>`",
432 label = "the service's own lock is missing",
433 note = "implement `Lock<{DB}>` for `{Self}`, or drop `lock` from `custom(..)` (by hand: `.own::<own::Lock>()` and `own::Lock` in `type Table`)"
434)]
435pub trait Lock<DB: Database>: InboxRow {
436 /// Tries the advisory lock on `key` for the session of `conn`, without waiting: `true` when it
437 /// took the lock. The session holds it until `Unlock` releases it or the session ends.
438 ///
439 /// # Errors
440 ///
441 /// The database's error; the claim fails, its connection closes, and the next claim waits one
442 /// second.
443 fn lock(
444 conn: &mut DB::Connection,
445 key: &str,
446 ) -> impl Future<Output = Result<bool, Error>> + Send;
447}
448
449/// Releases the advisory lock on a delivery's key: the `unlock` event of the advisory lock form,
450/// which a table names as its own beside [`Lock`].
451///
452/// A settlement calls it after its statement, on the connection that holds the delivery, and the
453/// connection goes back to the pool once it answered `true`. An unlock that answers `false` or
454/// fails closes the connection instead, which ends the session and its locks. A delivery dropped
455/// unsettled runs it before its connection closes, and `shutdown` runs it for every delivery in
456/// work.
457///
458/// # Examples
459///
460/// ```no_run
461/// # #[cfg(feature = "postgres")]
462/// # mod demo {
463/// use ruststream_sqlx::{Inbox, Lock, Unlock};
464/// use sqlx::{PgConnection, Postgres};
465///
466/// #[derive(Inbox, sqlx::FromRow)]
467/// #[inbox(table = "jobs", advisory_lock = "jobs-{id}", custom(lock, unlock))]
468/// pub struct Job {
469/// #[field(id)]
470/// id: i64,
471/// #[field(payload)]
472/// payload: Vec<u8>,
473/// }
474///
475/// // MSSQL releases with `sp_releaseapplock`; Postgres's session lock stands in for it here.
476/// impl Unlock<Postgres> for Job {
477/// async fn unlock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
478/// sqlx::query_scalar("SELECT pg_advisory_unlock(hashtext($1))")
479/// .bind(key)
480/// .fetch_one(conn)
481/// .await
482/// }
483/// }
484///
485/// impl Lock<Postgres> for Job {
486/// async fn lock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
487/// sqlx::query_scalar("SELECT pg_try_advisory_lock(hashtext($1))")
488/// .bind(key)
489/// .fetch_one(conn)
490/// .await
491/// }
492/// }
493/// # }
494/// # fn main() {}
495/// ```
496#[diagnostic::on_unimplemented(
497 message = "`{Self}` runs its own `unlock` (`custom(unlock)` in `#[inbox(..)]`, or `own::Unlock` in its `InboxSpec`) and does not implement `Unlock<{DB}>`",
498 label = "the service's own unlock is missing",
499 note = "implement `Unlock<{DB}>` for `{Self}`, or drop `unlock` from `custom(..)` (by hand: `.own::<own::Unlock>()` and `own::Unlock` in `type Table`)"
500)]
501pub trait Unlock<DB: Database>: InboxRow {
502 /// Releases the advisory lock on `key` the session of `conn` holds: `true` when the session
503 /// held it.
504 ///
505 /// # Errors
506 ///
507 /// The database's error; the connection closes, which ends the session and its locks.
508 fn unlock(
509 conn: &mut DB::Connection,
510 key: &str,
511 ) -> impl Future<Output = Result<bool, Error>> + Send;
512}
513
514/// Writes a published message into the table: the name, the bytes and the headers reach the
515/// service's SQL, which lays them out in its columns.
516///
517/// It has no default. With it, the struct's [`Repository`](crate::Repository) is a publisher and
518/// a route can lead names to it; without it, a mount that needs one does not compile, rather than
519/// publish into nothing. A message carrying a header the struct cannot hold byte for byte (any
520/// header, without a `#[field(headers)]` field; see [`HeaderColumn`](crate::HeaderColumn)) is
521/// refused before this runs.
522///
523/// # Examples
524///
525/// ```
526/// # #[cfg(feature = "postgres")]
527/// # mod demo {
528/// use ruststream::OutgoingMessage;
529/// use ruststream_sqlx::{Inbox, Publish};
530/// use sqlx::{PgConnection, Postgres};
531///
532/// #[derive(Inbox, sqlx::FromRow)]
533/// #[inbox(table = "email_jobs")]
534/// pub struct SendEmail {
535/// #[field(id, generated)]
536/// job_id: i64,
537/// #[field(group)]
538/// name: String,
539/// #[field(payload)]
540/// payload: Vec<u8>,
541/// }
542///
543/// impl Publish<Postgres> for SendEmail {
544/// async fn publish(
545/// conn: &mut PgConnection,
546/// message: &OutgoingMessage<'_>,
547/// ) -> Result<(), sqlx::Error> {
548/// sqlx::query("INSERT INTO email_jobs (name, payload) VALUES ($1, $2)")
549/// .bind(message.name())
550/// .bind(message.payload())
551/// .execute(conn)
552/// .await?;
553/// Ok(())
554/// }
555/// }
556/// # }
557/// # fn main() {}
558/// ```
559#[diagnostic::on_unimplemented(
560 message = "`{Self}` does not implement `Publish<{DB}>`, so nothing writes a published message into its table",
561 label = "no `Publish` for this row",
562 note = "implement `Publish<{DB}>` for `{Self}`: its SQL lays the name, the bytes and the headers out in the table"
563)]
564pub trait Publish<DB: Database>: InboxRow {
565 /// Writes `message` into the table.
566 ///
567 /// # Errors
568 ///
569 /// The database's error, which the publish returns.
570 fn publish(
571 conn: &mut DB::Connection,
572 message: &OutgoingMessage<'_>,
573 ) -> impl Future<Output = Result<(), Error>> + Send;
574}
575
576/// Inserts the row: every field except the `generated` ones, the queue's mechanics included, so
577/// a delayed task is a row whose `retry_after` lies ahead.
578///
579/// The derive implements it for the connection of each built-in dialect, `PgConnection` (feature
580/// `postgres`), `MySqlConnection` (feature `mysql`) and `SqliteConnection` (feature `sqlite`), with
581/// the statements built at compile time. A headers struct gets it from
582/// [`InboxHeaders`](derive@crate::InboxHeaders): the queue table's columns.
583/// A struct with a `#[sqlx(flatten)]` field gets none: the derive cannot see the nested struct's
584/// columns, so the service writes that insert itself.
585/// A table described by hand implements it over the text
586/// [`dialect::insert`](crate::dialect::insert) renders from its description, binding the columns
587/// in the description's order; [row mode](crate#row-mode) shows one.
588///
589/// # Examples
590///
591/// ```
592/// # #[cfg(feature = "postgres")]
593/// # mod demo {
594/// use ruststream::OutgoingMessage;
595/// use ruststream_sqlx::{Inbox, Insert, Publish};
596/// use sqlx::{PgConnection, Postgres};
597///
598/// #[derive(Inbox, sqlx::FromRow)]
599/// #[inbox(table = "email_jobs")]
600/// pub struct SendEmail {
601/// #[field(id, generated)]
602/// job_id: i64,
603/// #[field(group)]
604/// name: String,
605/// #[field(payload)]
606/// payload: Vec<u8>,
607/// }
608///
609/// impl Publish<Postgres> for SendEmail {
610/// async fn publish(
611/// conn: &mut PgConnection,
612/// message: &OutgoingMessage<'_>,
613/// ) -> Result<(), sqlx::Error> {
614/// let job = SendEmail {
615/// job_id: 0,
616/// name: message.name().to_owned(),
617/// payload: message.payload().to_vec(),
618/// };
619/// // `job_id` is generated: the database fills it in.
620/// job.insert(conn).await
621/// }
622/// }
623/// # }
624/// # fn main() {}
625/// ```
626#[diagnostic::on_unimplemented(
627 message = "`{Self}` has no generated insert for `{Connection}`",
628 label = "no `Insert` for this connection",
629 note = "the derive generates it for the connections of the built-in dialects (features \
630 `postgres`, `mysql` and `sqlite`), unless a `#[sqlx(flatten)]` field hides columns \
631 from it: write that insert in the service"
632)]
633pub trait Insert<Connection>: Sync {
634 /// Inserts the row on `conn`.
635 ///
636 /// # Errors
637 ///
638 /// The database's error.
639 ///
640 /// # Examples
641 ///
642 /// ```
643 /// # #[cfg(feature = "postgres")]
644 /// # mod demo {
645 /// use ruststream_sqlx::prelude::*;
646 /// use serde::Deserialize;
647 /// use sqlx::{PgPool, Postgres};
648 ///
649 /// #[derive(Inbox, sqlx::FromRow)]
650 /// #[inbox(table = "signup_jobs")]
651 /// pub struct Signup {
652 /// #[field(id, generated)]
653 /// id: i64,
654 /// #[field(payload)]
655 /// payload: Vec<u8>,
656 /// }
657 ///
658 /// /// A welcome mail to send, in row mode: the handler of `welcome_jobs` takes the row itself.
659 /// #[derive(Debug, Clone, Inbox, sqlx::FromRow)]
660 /// #[inbox(table = "welcome_jobs")]
661 /// pub struct Welcome {
662 /// #[field(id, generated)]
663 /// id: i64,
664 /// email: String,
665 /// }
666 ///
667 /// #[derive(Deserialize)]
668 /// struct Account {
669 /// email: String,
670 /// }
671 ///
672 /// // The welcome row commits with the signup's acknowledgement, or neither does.
673 /// #[subscriber(InboxQueue::<Signup>::new("signups"))]
674 /// async fn open_account(account: &Account, Ctx(mut tx): Ctx<keys::Tx<Postgres>>) -> HandlerOutcome {
675 /// let welcome = Welcome { id: 0, email: account.email.clone() };
676 /// match welcome.insert(&mut *tx).await {
677 /// Ok(()) => HandlerOutcome::ack(),
678 /// Err(_) => HandlerOutcome::retry(),
679 /// }
680 /// }
681 ///
682 /// pub fn app(pool: PgPool) -> RustStream {
683 /// RustStream::new(AppInfo::new("accounts", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
684 /// b.include(open_account.transactional());
685 /// })
686 /// }
687 /// # }
688 /// # fn main() {}
689 /// ```
690 fn insert<'c>(
691 &'c self,
692 conn: &'c mut Connection,
693 ) -> impl Future<Output = Result<(), Error>> + Send + 'c;
694}