ruststream_sqlx/outbox/events.rs
1//! One trait per event of an outbox record: what a service implements when it names the event as
2//! its own (`#[outbox(custom(..))]`, or `own` on its `OutboxSpec`), and `Publish`, which has no
3//! default.
4
5use std::future::Future;
6
7use ruststream::OutgoingMessage;
8use sqlx::{Database, Error};
9
10use super::OutboxRow;
11use super::database::Defaults;
12use super::dispatch::{AckBy, Declared, DiscardBy, FetchBy, HeadersOf, RecoverBy, RetryBy};
13use super::spec::{Declaration, OutboxTable};
14
15/// Creates the record of a message a handler publishes under a registered name, and returns its
16/// id, which the message then carries in [`OUTBOX_ID_HEADER`](super::OUTBOX_ID_HEADER).
17///
18/// It has no default: the service's statement lays the name, the payload and the headers out in
19/// its columns. A record type without it does not register.
20///
21/// # Examples
22///
23/// ```
24/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
25/// # mod demo {
26/// # use chrono::{DateTime, Utc};
27/// # use ruststream::OutgoingMessage;
28/// # use ruststream::memory::prelude::*;
29/// # use serde::{Deserialize, Serialize};
30/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
31/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
32/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
33/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
34/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
35/// use ruststream_sqlx::{Outbox, outbox};
36///
37/// #[derive(Outbox, sqlx::FromRow)]
38/// #[outbox(table = "outbox")]
39/// pub struct OrderOutbox {
40/// #[field(id)]
41/// id: i64,
42/// #[field(name)]
43/// name: String,
44/// #[field(payload)]
45/// payload: Vec<u8>,
46/// #[field(processed_at)]
47/// processed_at: Option<DateTime<Utc>>,
48/// }
49///
50/// impl outbox::Publish<Postgres> for OrderOutbox {
51/// async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
52/// sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
53/// .bind(msg.name())
54/// .bind(msg.payload())
55/// .fetch_one(conn)
56/// .await
57/// }
58/// }
59///
60/// pub fn app(pool: PgPool) -> impl App {
61/// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
62/// RustStream::new(AppInfo::new("orders", "0.1.0"))
63/// .layer(tracking.layer())
64/// .publish_layer(tracking.publish_layer())
65/// .with_broker(MemoryBroker::new(), |b| {
66/// b.include(place).out_reply(Publish);
67/// b.include(fulfil);
68/// b.after_startup(Publish, tracking.republish());
69/// })
70/// }
71/// # }
72/// # fn main() {}
73/// ```
74#[diagnostic::on_unimplemented(
75 message = "`{Self}` does not implement `outbox::Publish<{DB}>`, so nothing records what is published under its names",
76 label = "no `outbox::Publish` for this record",
77 note = "implement `outbox::Publish<{DB}>` for `{Self}`: its statement inserts the record and returns the id"
78)]
79pub trait Publish<DB: Database>: OutboxRow {
80 /// Inserts the record of `msg` and returns its id.
81 ///
82 /// # Errors
83 ///
84 /// The database's error; the message is not sent, and the publish fails with it.
85 fn publish(
86 conn: &mut DB::Connection,
87 msg: &OutgoingMessage<'_>,
88 ) -> impl Future<Output = Result<Self::Id, Error>> + Send;
89}
90
91/// Takes the record `id` into work when its message reaches a subscription.
92///
93/// The default selects the record while it is unprocessed. `None` means the record is taken or
94/// processed already: the handler does not run, and the delivery is acknowledged.
95///
96/// # Examples
97///
98/// ```
99/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
100/// # mod demo {
101/// # use chrono::{DateTime, Utc};
102/// # use ruststream::OutgoingMessage;
103/// # use ruststream::memory::prelude::*;
104/// # use serde::{Deserialize, Serialize};
105/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
106/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
107/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
108/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
109/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
110/// use ruststream_sqlx::{Outbox, outbox};
111///
112/// #[derive(Outbox, sqlx::FromRow)]
113/// #[outbox(table = "outbox", custom(fetch))]
114/// pub struct OrderOutbox {
115/// #[field(id)]
116/// id: i64,
117/// #[field(name)]
118/// name: String,
119/// #[field(payload)]
120/// payload: Vec<u8>,
121/// #[field(processed_at)]
122/// processed_at: Option<DateTime<Utc>>,
123/// }
124/// # impl outbox::Publish<Postgres> for OrderOutbox {
125/// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
126/// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
127/// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
128/// # }
129/// # }
130///
131/// // The service's fetch also stamps when the record was taken.
132/// impl outbox::Fetch<Postgres> for OrderOutbox {
133/// async fn fetch(conn: &mut PgConnection, id: &i64) -> sqlx::Result<Option<Self>> {
134/// sqlx::query_as(
135/// "UPDATE outbox SET taken_at = now() WHERE id = $1 AND processed_at IS NULL \
136/// RETURNING id, name, payload, processed_at",
137/// )
138/// .bind(id)
139/// .fetch_optional(conn)
140/// .await
141/// }
142/// }
143///
144/// pub fn app(pool: PgPool) -> impl App {
145/// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
146/// RustStream::new(AppInfo::new("orders", "0.1.0"))
147/// .layer(tracking.layer())
148/// .publish_layer(tracking.publish_layer())
149/// .with_broker(MemoryBroker::new(), |b| {
150/// b.include(place).out_reply(Publish);
151/// b.include(fulfil);
152/// b.after_startup(Publish, tracking.republish());
153/// })
154/// }
155/// # }
156/// # fn main() {}
157/// ```
158#[diagnostic::on_unimplemented(
159 message = "`{Self}` names its own fetch and does not implement `outbox::Fetch<{DB}>`",
160 label = "the service's own fetch is missing",
161 note = "implement `outbox::Fetch<{DB}>` for `{Self}`, or drop `fetch` from `#[outbox(custom(..))]` (or `.own::<own::Fetch>()` from its `OutboxSpec`)"
162)]
163pub trait Fetch<DB: Database>: OutboxRow {
164 /// Takes the record `id` into work, or `None` when it is taken or processed already.
165 ///
166 /// # Errors
167 ///
168 /// The database's error; the handler does not run, and the delivery is retried.
169 fn fetch(
170 conn: &mut DB::Connection,
171 id: &Self::Id,
172 ) -> impl Future<Output = Result<Option<Self>, Error>> + Send;
173}
174
175macro_rules! outcome_event {
176 (
177 $(#[$doc:meta])* $trait:ident, $method:ident, $event:literal,
178 message = $message:literal, note = $note:literal
179 ) => {
180 $(#[$doc])*
181 #[diagnostic::on_unimplemented(
182 message = $message,
183 label = "the service's own event is missing",
184 note = $note
185 )]
186 pub trait $trait<DB: Database>: OutboxRow {
187 #[doc = concat!("Runs the `", $event, "` event for the record `id` after its handler finished.")]
188 #[doc = ""]
189 #[doc = "# Errors"]
190 #[doc = ""]
191 #[doc = "The database's error; the record stays unprocessed, and the next startup publishes it again."]
192 fn $method(
193 conn: &mut DB::Connection,
194 id: &Self::Id,
195 ) -> impl Future<Output = Result<(), Error>> + Send;
196 }
197 };
198}
199
200outcome_event!(
201 /// Marks the record processed once its handler acknowledged the message.
202 ///
203 /// The default sets `processed_at` from the database's clock, or deletes the record when the
204 /// struct has no `processed_at` field.
205 ///
206 /// # Examples
207 ///
208 /// ```
209 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
210 /// # mod demo {
211 /// # use chrono::{DateTime, Utc};
212 /// # use ruststream::OutgoingMessage;
213 /// # use ruststream::memory::prelude::*;
214 /// # use serde::{Deserialize, Serialize};
215 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
216 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
217 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
218 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
219 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
220 /// use ruststream_sqlx::{Outbox, outbox};
221 ///
222 /// #[derive(Outbox, sqlx::FromRow)]
223 /// #[outbox(table = "outbox", custom(ack))]
224 /// pub struct OrderOutbox {
225 /// #[field(id)]
226 /// id: i64,
227 /// #[field(name)]
228 /// name: String,
229 /// #[field(payload)]
230 /// payload: Vec<u8>,
231 /// #[field(processed_at)]
232 /// processed_at: Option<DateTime<Utc>>,
233 /// }
234 /// # impl outbox::Publish<Postgres> for OrderOutbox {
235 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
236 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
237 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
238 /// # }
239 /// # }
240 ///
241 /// // A processed record moves to the archive.
242 /// impl outbox::Ack<Postgres> for OrderOutbox {
243 /// async fn ack(conn: &mut PgConnection, id: &i64) -> sqlx::Result<()> {
244 /// sqlx::query(
245 /// "WITH done AS (DELETE FROM outbox WHERE id = $1 RETURNING id, name, payload) \
246 /// INSERT INTO outbox_archive (id, name, payload) SELECT id, name, payload FROM done",
247 /// )
248 /// .bind(id)
249 /// .execute(conn)
250 /// .await?;
251 /// Ok(())
252 /// }
253 /// }
254 ///
255 /// pub fn app(pool: PgPool) -> impl App {
256 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
257 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
258 /// .layer(tracking.layer())
259 /// .publish_layer(tracking.publish_layer())
260 /// .with_broker(MemoryBroker::new(), |b| {
261 /// b.include(place).out_reply(Publish);
262 /// b.include(fulfil);
263 /// b.after_startup(Publish, tracking.republish());
264 /// })
265 /// }
266 /// # }
267 /// # fn main() {}
268 /// ```
269 Ack, ack, "ack",
270 message = "`{Self}` names its own ack and does not implement `outbox::Ack<{DB}>`",
271 note = "implement `outbox::Ack<{DB}>` for `{Self}`, or drop `ack` from `#[outbox(custom(..))]` (or `.own::<own::Ack>()` from its `OutboxSpec`)"
272);
273
274outcome_event!(
275 /// Runs when the handler asked for the message again.
276 ///
277 /// The default leaves the record unprocessed and runs no statement, so the next startup
278 /// publishes it again.
279 ///
280 /// # Examples
281 ///
282 /// ```
283 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
284 /// # mod demo {
285 /// # use chrono::{DateTime, Utc};
286 /// # use ruststream::OutgoingMessage;
287 /// # use ruststream::memory::prelude::*;
288 /// # use serde::{Deserialize, Serialize};
289 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
290 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
291 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
292 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
293 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
294 /// use ruststream_sqlx::{Outbox, outbox};
295 ///
296 /// #[derive(Outbox, sqlx::FromRow)]
297 /// #[outbox(table = "outbox", custom(retry))]
298 /// pub struct OrderOutbox {
299 /// #[field(id)]
300 /// id: i64,
301 /// #[field(name)]
302 /// name: String,
303 /// #[field(payload)]
304 /// payload: Vec<u8>,
305 /// #[field(processed_at)]
306 /// processed_at: Option<DateTime<Utc>>,
307 /// }
308 /// # impl outbox::Publish<Postgres> for OrderOutbox {
309 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
310 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
311 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
312 /// # }
313 /// # }
314 ///
315 /// // The service counts how often a consumer asked for a record again.
316 /// impl outbox::Retry<Postgres> for OrderOutbox {
317 /// async fn retry(conn: &mut PgConnection, id: &i64) -> sqlx::Result<()> {
318 /// sqlx::query("UPDATE outbox SET retries = retries + 1 WHERE id = $1")
319 /// .bind(id)
320 /// .execute(conn)
321 /// .await?;
322 /// Ok(())
323 /// }
324 /// }
325 ///
326 /// pub fn app(pool: PgPool) -> impl App {
327 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
328 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
329 /// .layer(tracking.layer())
330 /// .publish_layer(tracking.publish_layer())
331 /// .with_broker(MemoryBroker::new(), |b| {
332 /// b.include(place).out_reply(Publish);
333 /// b.include(fulfil);
334 /// b.after_startup(Publish, tracking.republish());
335 /// })
336 /// }
337 /// # }
338 /// # fn main() {}
339 /// ```
340 Retry, retry, "retry",
341 message = "`{Self}` names its own retry and does not implement `outbox::Retry<{DB}>`",
342 note = "implement `outbox::Retry<{DB}>` for `{Self}`, or drop `retry` from `#[outbox(custom(..))]` (or `.own::<own::Retry>()` from its `OutboxSpec`)"
343);
344
345outcome_event!(
346 /// Runs when the handler dropped the message.
347 ///
348 /// The default marks the record processed as [`Ack`] does: a dropped message is not sent
349 /// again.
350 ///
351 /// # Examples
352 ///
353 /// ```
354 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
355 /// # mod demo {
356 /// # use chrono::{DateTime, Utc};
357 /// # use ruststream::OutgoingMessage;
358 /// # use ruststream::memory::prelude::*;
359 /// # use serde::{Deserialize, Serialize};
360 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
361 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
362 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
363 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
364 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
365 /// use ruststream_sqlx::{Outbox, outbox};
366 ///
367 /// #[derive(Outbox, sqlx::FromRow)]
368 /// #[outbox(table = "outbox", custom(discard))]
369 /// pub struct OrderOutbox {
370 /// #[field(id)]
371 /// id: i64,
372 /// #[field(name)]
373 /// name: String,
374 /// #[field(payload)]
375 /// payload: Vec<u8>,
376 /// #[field(processed_at)]
377 /// processed_at: Option<DateTime<Utc>>,
378 /// }
379 /// # impl outbox::Publish<Postgres> for OrderOutbox {
380 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
381 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
382 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
383 /// # }
384 /// # }
385 ///
386 /// // A dropped record is marked apart from a processed one.
387 /// impl outbox::Discard<Postgres> for OrderOutbox {
388 /// async fn discard(conn: &mut PgConnection, id: &i64) -> sqlx::Result<()> {
389 /// sqlx::query("UPDATE outbox SET processed_at = now(), dropped = true WHERE id = $1")
390 /// .bind(id)
391 /// .execute(conn)
392 /// .await?;
393 /// Ok(())
394 /// }
395 /// }
396 ///
397 /// pub fn app(pool: PgPool) -> impl App {
398 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
399 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
400 /// .layer(tracking.layer())
401 /// .publish_layer(tracking.publish_layer())
402 /// .with_broker(MemoryBroker::new(), |b| {
403 /// b.include(place).out_reply(Publish);
404 /// b.include(fulfil);
405 /// b.after_startup(Publish, tracking.republish());
406 /// })
407 /// }
408 /// # }
409 /// # fn main() {}
410 /// ```
411 Discard, discard, "discard",
412 message = "`{Self}` names its own discard and does not implement `outbox::Discard<{DB}>`",
413 note = "implement `outbox::Discard<{DB}>` for `{Self}`, or drop `discard` from `#[outbox(custom(..))]` (or `.own::<own::Discard>()` from its `OutboxSpec`)"
414);
415
416/// Selects the unprocessed records of one name, for the republish at startup.
417///
418/// The default selects the records of `name` whose `processed_at` is `NULL`, or every record of
419/// `name` when the struct has no `processed_at` field.
420///
421/// # Examples
422///
423/// ```
424/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
425/// # mod demo {
426/// # use chrono::{DateTime, Utc};
427/// # use ruststream::OutgoingMessage;
428/// # use ruststream::memory::prelude::*;
429/// # use serde::{Deserialize, Serialize};
430/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
431/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
432/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
433/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
434/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
435/// use ruststream_sqlx::{Outbox, outbox};
436///
437/// #[derive(Outbox, sqlx::FromRow)]
438/// #[outbox(table = "outbox", custom(recover))]
439/// pub struct OrderOutbox {
440/// #[field(id)]
441/// id: i64,
442/// #[field(name)]
443/// name: String,
444/// #[field(payload)]
445/// payload: Vec<u8>,
446/// #[field(processed_at)]
447/// processed_at: Option<DateTime<Utc>>,
448/// }
449/// # impl outbox::Publish<Postgres> for OrderOutbox {
450/// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
451/// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
452/// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
453/// # }
454/// # }
455///
456/// // A record published in the last minute may still be in flight; the next startup takes it.
457/// impl outbox::Recover<Postgres> for OrderOutbox {
458/// async fn recover(conn: &mut PgConnection, name: &str) -> sqlx::Result<Vec<Self>> {
459/// sqlx::query_as(
460/// "SELECT id, name, payload, processed_at FROM outbox \
461/// WHERE name = $1 AND processed_at IS NULL AND created_at < now() - interval '1 minute'",
462/// )
463/// .bind(name)
464/// .fetch_all(conn)
465/// .await
466/// }
467/// }
468///
469/// pub fn app(pool: PgPool) -> impl App {
470/// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
471/// RustStream::new(AppInfo::new("orders", "0.1.0"))
472/// .layer(tracking.layer())
473/// .publish_layer(tracking.publish_layer())
474/// .with_broker(MemoryBroker::new(), |b| {
475/// b.include(place).out_reply(Publish);
476/// b.include(fulfil);
477/// b.after_startup(Publish, tracking.republish());
478/// })
479/// }
480/// # }
481/// # fn main() {}
482/// ```
483#[diagnostic::on_unimplemented(
484 message = "`{Self}` names its own recovery and does not implement `outbox::Recover<{DB}>`",
485 label = "the service's own recovery is missing",
486 note = "implement `outbox::Recover<{DB}>` for `{Self}`, or drop `recover` from `#[outbox(custom(..))]` (or `.own::<own::Recover>()` from its `OutboxSpec`)"
487)]
488pub trait Recover<DB: Database>: OutboxRow {
489 /// The unprocessed records published under `name`.
490 ///
491 /// # Errors
492 ///
493 /// The database's error; startup fails with it.
494 fn recover(
495 conn: &mut DB::Connection,
496 name: &str,
497 ) -> impl Future<Output = Result<Vec<Self>, Error>> + Send;
498}
499
500/// Every event of a record type on `DB`, each dispatched by its slot in the record's settings:
501/// what registering a record requires. Implemented for every [`OutboxTable`] that implements
502/// [`Publish`] and every event its settings name as its own. Machinery.
503// No message of its own: rustc then reports the unmet bound underneath (the missing `Publish`, the
504// missing event of the record's own), with that trait's message.
505#[doc(hidden)]
506pub trait Tracked<DB: Database>: Publish<DB> {
507 /// The record's headers, moved out of it; empty without a headers column.
508 fn take_headers(&mut self) -> ruststream::HeaderMap;
509
510 /// Takes the record `id` into work.
511 fn fetch_record<'c>(
512 conn: &'c mut DB::Connection,
513 id: &'c Self::Id,
514 defaults: &'c Defaults,
515 ) -> impl Future<Output = Result<Option<Self>, Error>> + Send + 'c;
516
517 /// Marks the record `id` acknowledged.
518 fn ack_record<'c>(
519 conn: &'c mut DB::Connection,
520 id: &'c Self::Id,
521 defaults: &'c Defaults,
522 ) -> impl Future<Output = Result<(), Error>> + Send + 'c;
523
524 /// Runs the retry of the record `id`.
525 fn retry_record<'c>(
526 conn: &'c mut DB::Connection,
527 id: &'c Self::Id,
528 defaults: &'c Defaults,
529 ) -> impl Future<Output = Result<(), Error>> + Send + 'c;
530
531 /// Marks the record `id` dropped.
532 fn discard_record<'c>(
533 conn: &'c mut DB::Connection,
534 id: &'c Self::Id,
535 defaults: &'c Defaults,
536 ) -> impl Future<Output = Result<(), Error>> + Send + 'c;
537
538 /// The unprocessed records published under `name`.
539 fn recover_records<'c>(
540 conn: &'c mut DB::Connection,
541 name: &'c str,
542 defaults: &'c Defaults,
543 ) -> impl Future<Output = Result<Vec<Self>, Error>> + Send + 'c;
544}
545
546impl<DB, Record> Tracked<DB> for Record
547where
548 DB: Database,
549 Record: OutboxTable + Publish<DB>,
550 <Declared<Record> as Declaration>::Headers: HeadersOf<Record>,
551 <Declared<Record> as Declaration>::OwnFetch: FetchBy<DB, Record>,
552 <Declared<Record> as Declaration>::OwnAck: AckBy<DB, Record>,
553 <Declared<Record> as Declaration>::OwnRetry: RetryBy<DB, Record>,
554 <Declared<Record> as Declaration>::OwnDiscard: DiscardBy<DB, Record>,
555 <Declared<Record> as Declaration>::OwnRecover: RecoverBy<DB, Record>,
556{
557 #[inline]
558 fn take_headers(&mut self) -> ruststream::HeaderMap {
559 <<Declared<Record> as Declaration>::Headers as HeadersOf<Record>>::take(self)
560 }
561
562 fn fetch_record<'c>(
563 conn: &'c mut DB::Connection,
564 id: &'c <Record as OutboxTable>::Id,
565 defaults: &'c Defaults,
566 ) -> impl Future<Output = Result<Option<Self>, Error>> + Send + 'c {
567 <<Declared<Record> as Declaration>::OwnFetch as FetchBy<DB, Record>>::fetch(
568 conn, id, defaults,
569 )
570 }
571
572 fn ack_record<'c>(
573 conn: &'c mut DB::Connection,
574 id: &'c <Record as OutboxTable>::Id,
575 defaults: &'c Defaults,
576 ) -> impl Future<Output = Result<(), Error>> + Send + 'c {
577 <<Declared<Record> as Declaration>::OwnAck as AckBy<DB, Record>>::ack(conn, id, defaults)
578 }
579
580 fn retry_record<'c>(
581 conn: &'c mut DB::Connection,
582 id: &'c <Record as OutboxTable>::Id,
583 defaults: &'c Defaults,
584 ) -> impl Future<Output = Result<(), Error>> + Send + 'c {
585 <<Declared<Record> as Declaration>::OwnRetry as RetryBy<DB, Record>>::retry(
586 conn, id, defaults,
587 )
588 }
589
590 fn discard_record<'c>(
591 conn: &'c mut DB::Connection,
592 id: &'c <Record as OutboxTable>::Id,
593 defaults: &'c Defaults,
594 ) -> impl Future<Output = Result<(), Error>> + Send + 'c {
595 <<Declared<Record> as Declaration>::OwnDiscard as DiscardBy<DB, Record>>::discard(
596 conn, id, defaults,
597 )
598 }
599
600 fn recover_records<'c>(
601 conn: &'c mut DB::Connection,
602 name: &'c str,
603 defaults: &'c Defaults,
604 ) -> impl Future<Output = Result<Vec<Self>, Error>> + Send + 'c {
605 <<Declared<Record> as Declaration>::OwnRecover as RecoverBy<DB, Record>>::recover(
606 conn, name, defaults,
607 )
608 }
609}