ruststream_sqlx/inbox/
manual.rs

1//! A queue table described by hand: the roles a delivery reads off a row, one small trait each,
2//! and the row's whole contract with the crate, implemented once for every
3//! [`InboxTable`](crate::InboxTable) and generic over the database.
4//!
5//! The contract reads the table's settings from the type of its description: each setting is a
6//! type, and each trait of [`axes`] and [`dispatch`] picks the code for it, so the derived table
7//! and the table described by hand call the same helpers with the same type arguments.
8
9mod axes;
10mod check;
11mod dispatch;
12
13use std::future::Future;
14use std::time::Duration;
15
16use ruststream::HeaderMap;
17use ruststream_sqlx_dialect::{Param, Role, TableSpec};
18use sqlx::{Decode, Encode, Error, FromRow, Type};
19
20pub use axes::{
21    AttemptAxis, ClockAxis, FormAxis, FormBind, HeadersAxis, KeyAxis, LeaseAxis, ManualHeaders,
22    MessageAxis, Named, OpeningAxis, TimeAxis,
23};
24pub use dispatch::{
25    AckAxis, ClaimAxes, DeadLetterAxis, DiscardAxis, ExtendAxis, LockAxis, RetryAfterAxis,
26    RetryAxis, UnlockAxis,
27};
28
29use super::columns::{AttemptColumn, KeyColumn};
30use super::database::QueueDatabase;
31use super::engine::{
32    self, Claimed, Claiming, Event, Events, Leasing, Now, Settled, Settling, Shape, Values,
33};
34use super::form::advisory::events::{self as advisory, Candidates};
35use super::headers::{Assembled, HeaderField};
36use super::named::kinds::{Kinds, KindsOf};
37use super::queue::Queue;
38use super::spec::Declaration;
39use super::time::LeaseRow;
40use super::{InboxRow, InboxSpec, InboxTable, QueueRow};
41
42/// The row's partition key, which a table described by hand reads off its row where its
43/// description sets [`InboxSpec::partition_key`].
44///
45/// # Examples
46///
47/// ```
48/// # #[cfg(all(feature = "sqlite", feature = "chrono", feature = "json"))]
49/// # mod demo {
50/// use chrono::{DateTime, Utc};
51/// use ruststream::runtime::{Input, SoloCarried};
52/// use ruststream_sqlx::dialect::Column;
53/// use ruststream_sqlx::prelude::*;
54/// use ruststream_sqlx::spec::{Key, Lease};
55/// use ruststream_sqlx::{InboxSpec, InboxTable, KeyRow};
56/// use sqlx::{Sqlite, SqlitePool};
57///
58/// #[derive(Debug, Clone, sqlx::FromRow)]
59/// pub struct Invoice {
60///     id: i64,
61///     customer: String,
62///     total: i64,
63/// }
64///
65/// impl InboxTable for Invoice {
66///     type Id = i64;
67///     type Table = InboxSpec<(Lease<DateTime<Utc>>, Key)>;
68///     const TABLE: Self::Table = InboxSpec::new("invoices", Column::new("id").generated())
69///         .lease(Column::new("locked_until"))
70///         .partition_key(Column::new("customer"))
71///         .data(&[Column::new("total")]);
72///
73///     fn id(&self) -> &i64 {
74///         &self.id
75///     }
76/// }
77///
78/// // Row mode: the handler takes the row itself.
79/// impl Input for Invoice {
80///     type Axis = SoloCarried<Self>;
81/// }
82///
83/// impl KeyRow for Invoice {
84///     type Key = String;
85///
86///     fn partition_key(&self) -> &String {
87///         &self.customer
88///     }
89/// }
90///
91/// // One customer's invoices settle in order, each customer on a lane of its own.
92/// #[subscriber(InboxQueue::<Invoice>::new("invoices"))]
93/// async fn bill(invoice: &Invoice) -> HandlerOutcome {
94///     if invoice.total >= 0 {
95///         HandlerOutcome::ack()
96///     } else {
97///         HandlerOutcome::drop()
98///     }
99/// }
100///
101/// pub fn app(pool: SqlitePool) -> RustStream {
102///     RustStream::new(AppInfo::new("billing", "1.0.0")).with_broker(
103///         SqlxBroker::<Sqlite>::new(pool),
104///         |b| {
105///             b.include(bill);
106///         },
107///     )
108/// }
109/// # }
110/// # fn main() {}
111/// ```
112#[diagnostic::on_unimplemented(
113    message = "`{Self}` sets a `partition_key` column and does not read it off its row",
114    label = "no `KeyRow` for this row",
115    note = "implement `KeyRow` for `{Self}`, or drop `.partition_key(..)` and `Key` from its \
116            description"
117)]
118pub trait KeyRow {
119    /// The key column's type: text or bytes, optional or not.
120    type Key: KeyColumn + 'static;
121
122    /// The key field.
123    fn partition_key(&self) -> &Self::Key;
124}
125
126/// The row's attempt, which a table described by hand reads off its row where its description
127/// sets [`InboxSpec::attempt`].
128///
129/// The column's type is part of the trait, so a row that does not decode still reports its
130/// attempt, read alone.
131///
132/// # Examples
133///
134/// ```
135/// # #[cfg(all(feature = "sqlite", feature = "chrono", feature = "json"))]
136/// # mod demo {
137/// use chrono::{DateTime, Utc};
138/// use ruststream_sqlx::dialect::Column;
139/// use ruststream_sqlx::prelude::*;
140/// use ruststream_sqlx::spec::{Attempt, Lease, Payload};
141/// use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable, PayloadRow};
142/// use serde::Deserialize;
143/// use sqlx::{Sqlite, SqlitePool};
144///
145/// #[derive(sqlx::FromRow)]
146/// pub struct EmailJob {
147///     job_id: i64,
148///     attempt: i16,
149///     payload: Vec<u8>,
150/// }
151///
152/// impl InboxTable for EmailJob {
153///     type Id = i64;
154///     type Table = InboxSpec<(Lease<DateTime<Utc>>, Attempt, Payload)>;
155///     const TABLE: Self::Table = InboxSpec::new("email_jobs", Column::new("job_id").generated())
156///         .lease(Column::new("locked_until"))
157///         .group(Column::new("name"))
158///         .attempt(Column::new("attempt").generated())
159///         .payload(Column::new("payload"));
160///
161///     fn id(&self) -> &i64 {
162///         &self.job_id
163///     }
164/// }
165///
166/// impl PayloadRow for EmailJob {
167///     type Column = Vec<u8>;
168///
169///     fn payload(&self) -> &[u8] {
170///         &self.payload
171///     }
172/// }
173///
174/// impl AttemptRow for EmailJob {
175///     type Attempt = i16;
176///
177///     fn attempt(&self) -> &i16 {
178///         &self.attempt
179///     }
180/// }
181///
182/// #[derive(Deserialize)]
183/// struct Email {
184///     to: String,
185/// }
186///
187/// // The third attempt gives up on an address the first two could not reach.
188/// #[subscriber(InboxQueue::<EmailJob>::new("emails"))]
189/// async fn send(email: &Email, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
190///     match attempt {
191///         Some(3..) => HandlerOutcome::drop(),
192///         _ if email.to.contains('@') => HandlerOutcome::ack(),
193///         _ => HandlerOutcome::retry(),
194///     }
195/// }
196///
197/// pub fn app(pool: SqlitePool) -> RustStream {
198///     RustStream::new(AppInfo::new("mailer", "1.0.0")).with_broker(
199///         SqlxBroker::<Sqlite>::new(pool),
200///         |b| {
201///             b.include(send);
202///         },
203///     )
204/// }
205/// # }
206/// # fn main() {}
207/// ```
208#[diagnostic::on_unimplemented(
209    message = "`{Self}` sets an `attempt` column and does not read it off its row",
210    label = "no `AttemptRow` for this row",
211    note = "implement `AttemptRow` for `{Self}`, or drop `.attempt(..)` and `Attempt` from its \
212            description"
213)]
214pub trait AttemptRow {
215    /// The attempt column's type: a signed integer.
216    type Attempt: AttemptColumn + 'static;
217
218    /// The attempt field.
219    fn attempt(&self) -> &Self::Attempt;
220}
221
222/// Header fields of the row: where its description sets [`InboxSpec::header_fields`], the
223/// delivery builds its header map from them on its first read.
224///
225/// # Examples
226///
227/// ```
228/// # #[cfg(all(feature = "sqlite", feature = "chrono", feature = "json"))]
229/// # mod demo {
230/// use chrono::{DateTime, Utc};
231/// use ruststream::HeaderMap;
232/// use ruststream::runtime::{Input, SoloCarried};
233/// use ruststream_sqlx::dialect::Column;
234/// use ruststream_sqlx::prelude::*;
235/// use ruststream_sqlx::spec::{self, Lease};
236/// use ruststream_sqlx::{HeaderFields, InboxSpec, InboxTable, put_header};
237/// use sqlx::{Sqlite, SqlitePool};
238///
239/// #[derive(Debug, Clone, sqlx::FromRow)]
240/// pub struct OrderJob {
241///     job_id: i64,
242///     tenant: String,
243///     trace: Option<String>,
244///     note: Option<String>,
245/// }
246///
247/// impl InboxTable for OrderJob {
248///     type Id = i64;
249///     type Table = InboxSpec<(Lease<DateTime<Utc>>, spec::HeaderFields)>;
250///     const TABLE: Self::Table = InboxSpec::new("order_jobs", Column::new("job_id").generated())
251///         .lease(Column::new("locked_until"))
252///         .group(Column::new("name"))
253///         .data(&[Column::new("tenant"), Column::new("trace"), Column::new("note")])
254///         .header_fields();
255///
256///     fn id(&self) -> &i64 {
257///         &self.job_id
258///     }
259/// }
260///
261/// impl Input for OrderJob {
262///     type Axis = SoloCarried<Self>;
263/// }
264///
265/// impl HeaderFields for OrderJob {
266///     const NAMES: &'static [&'static str] = &["tenant", "trace"];
267///
268///     fn header_map(&self) -> HeaderMap {
269///         let mut headers = HeaderMap::with_capacity(Self::NAMES.len());
270///         put_header(&mut headers, "tenant", &self.tenant);
271///         put_header(&mut headers, "trace", &self.trace);
272///         headers
273///     }
274/// }
275///
276/// // The tenant's header reads like any other; the row holds the note.
277/// #[subscriber(InboxQueue::<OrderJob>::new("orders"))]
278/// async fn ship(job: &OrderJob, ctx: &mut Context<'_>) -> HandlerOutcome {
279///     match (ctx.headers().get_str("tenant"), job.note.as_deref()) {
280///         (Some(_), Some("fragile")) => HandlerOutcome::retry(),
281///         _ => HandlerOutcome::ack(),
282///     }
283/// }
284///
285/// pub fn app(pool: SqlitePool) -> RustStream {
286///     RustStream::new(AppInfo::new("shop", "1.0.0")).with_broker(
287///         SqlxBroker::<Sqlite>::new(pool),
288///         |b| {
289///             b.include(ship);
290///         },
291///     )
292/// }
293/// # }
294/// # fn main() {}
295/// ```
296#[diagnostic::on_unimplemented(
297    message = "`{Self}` builds its headers from fields and does not say which",
298    label = "no `HeaderFields` for this row",
299    note = "implement `HeaderFields` for `{Self}`, or drop `.header_fields()` and \
300            `spec::HeaderFields` from its description"
301)]
302pub trait HeaderFields {
303    /// The headers the fields hold, by name; a publish that carries another is refused.
304    const NAMES: &'static [&'static str];
305
306    /// The header map: each field under its name, through [`put_header`].
307    fn header_map(&self) -> HeaderMap;
308}
309
310/// Puts `field` into `headers` under `name`, the way a field of a struct deriving
311/// [`InboxHeaders`](crate::InboxHeaders) becomes a header; a field without a value leaves the
312/// header out.
313///
314/// # Examples
315///
316/// ```
317/// use ruststream::HeaderMap;
318/// use ruststream_sqlx::put_header;
319///
320/// let tenant = String::from("acme");
321/// let trace: Option<String> = None;
322/// let mut headers = HeaderMap::with_capacity(2);
323/// put_header(&mut headers, "tenant", &tenant);
324/// put_header(&mut headers, "trace", &trace);
325/// assert_eq!(headers.get_str("tenant"), Some("acme"));
326/// assert_eq!(headers.get_str("trace"), None);
327/// ```
328pub fn put_header(headers: &mut HeaderMap, name: &'static str, field: &impl HeaderField) {
329    super::headers::put_header(headers, name, field);
330}
331
332impl<Row: HeaderFields> Assembled for Row {
333    fn header_map(&self) -> HeaderMap {
334        HeaderFields::header_map(self)
335    }
336}
337
338impl<Row, Settings> QueueRow for Row
339where
340    Row: InboxTable<Table = InboxSpec<Settings>>,
341    Settings: Declaration,
342    Settings::Message: MessageAxis<Row>,
343{
344    type Id = <Row as InboxTable>::Id;
345    type Lane = <Settings::Message as MessageAxis<Row>>::Lane;
346}
347
348impl<Row, Settings> InboxRow for Row
349where
350    Row: InboxTable<Table = InboxSpec<Settings>>,
351    Settings: Declaration,
352    Settings::Message: MessageAxis<Row>,
353    Settings::Form: FormAxis,
354    Settings::Opening: OpeningAxis,
355{
356    const SPEC: TableSpec<'static> = check::checked(&Row::TABLE.spec());
357    type Form = <Settings::Form as FormAxis>::Form;
358    type Opening = <Settings::Opening as OpeningAxis>::Opening;
359}
360
361// Reported as the row without `LeaseRow`, whose message names the fix, not as the form setting
362// it reads.
363#[diagnostic::do_not_recommend]
364impl<Row, Settings> LeaseRow for Row
365where
366    Row: InboxRow + InboxTable<Table = InboxSpec<Settings>>,
367    Settings: Declaration,
368    Settings::Form: LeaseAxis,
369{
370    type Lease = <Settings::Form as LeaseAxis>::Time;
371}
372
373/// The settings `Events` reads as constants.
374pub trait Axes {
375    /// Which events the service writes itself.
376    const SHAPE: Shape;
377    /// Whether the service writes any event itself, which a by-name subscription cannot run.
378    const OWN: bool;
379}
380
381impl<Settings> Axes for InboxSpec<Settings>
382where
383    Settings: Declaration,
384    Settings::OwnClaim: Named,
385    Settings::OwnFetch: Named,
386    Settings::OwnAck: Named,
387    Settings::OwnRetry: Named,
388    Settings::OwnRetryAfter: Named,
389    Settings::OwnDiscard: Named,
390    Settings::OwnDeadLetter: Named,
391    Settings::OwnExtend: Named,
392    Settings::OwnLock: Named,
393    Settings::OwnUnlock: Named,
394{
395    const SHAPE: Shape = Shape {
396        custom_claim: Settings::OwnClaim::NAMED,
397        custom_fetch: Settings::OwnFetch::NAMED,
398        custom_ack: Settings::OwnAck::NAMED,
399        custom_retry: Settings::OwnRetry::NAMED,
400        custom_retry_after: Settings::OwnRetryAfter::NAMED,
401        custom_discard: Settings::OwnDiscard::NAMED,
402        custom_dead_letter: Settings::OwnDeadLetter::NAMED,
403        custom_extend: Settings::OwnExtend::NAMED,
404        custom_lock: Settings::OwnLock::NAMED,
405        custom_unlock: Settings::OwnUnlock::NAMED,
406    };
407    const OWN: bool = Settings::OwnClaim::NAMED
408        || Settings::OwnFetch::NAMED
409        || Settings::OwnAck::NAMED
410        || Settings::OwnRetry::NAMED
411        || Settings::OwnRetryAfter::NAMED
412        || Settings::OwnDiscard::NAMED
413        || Settings::OwnDeadLetter::NAMED
414        || Settings::OwnExtend::NAMED
415        || Settings::OwnLock::NAMED
416        || Settings::OwnUnlock::NAMED;
417}
418
419type Own<Settings> = (
420    <Settings as Declaration>::OwnClaim,
421    <Settings as Declaration>::OwnFetch,
422);
423type Token<Settings> = <<Settings as Declaration>::Form as FormAxis>::Token;
424type Source<Settings> = <<Settings as Declaration>::Clock as ClockAxis>::Source;
425
426impl<DB, Row, Settings> Events<DB> for Row
427where
428    DB: QueueDatabase,
429    Row: InboxTable<Table = InboxSpec<Settings>> + for<'r> FromRow<'r, DB::Row> + Unpin,
430    <Row as InboxTable>::Id: for<'q> Encode<'q, DB> + for<'r> Decode<'r, DB> + Type<DB>,
431    Settings: Declaration,
432    InboxSpec<Settings>: Axes,
433    Settings::Message: MessageAxis<Row>,
434    Settings::Form: FormBind<DB>,
435    Settings::Key: KeyAxis<Row>,
436    Settings::Attempt: AttemptAxis<DB, Row>,
437    Settings::Headers: HeadersAxis<Row> + 'static,
438    Settings::RetryAfter: TimeAxis<DB>,
439    Settings::ProcessedAt: TimeAxis<DB>,
440    Settings::Clock: ClockAxis,
441    Settings::Opening: OpeningAxis,
442    Own<Settings>: ClaimAxes<DB, Row>,
443    Settings::OwnAck: AckAxis<DB, Row>,
444    Settings::OwnRetry: RetryAxis<DB, Row>,
445    Settings::OwnRetryAfter: RetryAfterAxis<DB, Row>,
446    Settings::OwnDiscard: DiscardAxis<DB, Row>,
447    Settings::OwnDeadLetter: DeadLetterAxis<DB, Row>,
448    Settings::OwnExtend: ExtendAxis<DB, Row, Token<Settings>>,
449    Settings::OwnLock: LockAxis<DB, Row>,
450    Settings::OwnUnlock: UnlockAxis<DB, Row>,
451{
452    const SHAPE: Shape = <InboxSpec<Settings> as Axes>::SHAPE;
453
454    type Token = Token<Settings>;
455
456    type Ids = <Own<Settings> as ClaimAxes<DB, Row>>::Ids;
457
458    type Headers = ManualHeaders<Settings::Headers, Row>;
459
460    fn kinds() -> Option<Kinds> {
461        // A by-name subscription runs the crate's own events only.
462        if <InboxSpec<Settings> as Axes>::OWN {
463            return None;
464        }
465        let kinds = KindsOf::new::<<Row as InboxTable>::Id, Source<Settings>>();
466        let kinds = <Settings::Message as MessageAxis<Row>>::kinds(kinds);
467        let kinds = <Settings::Headers as HeadersAxis<Row>>::kinds(kinds);
468        let kinds = <Settings::Key as KeyAxis<Row>>::kinds(kinds);
469        let kinds = <Settings::Attempt as AttemptAxis<DB, Row>>::kinds(kinds);
470        let kinds = <Settings::RetryAfter as TimeAxis<DB>>::kinds(kinds, true);
471        let kinds = <Settings::ProcessedAt as TimeAxis<DB>>::kinds(kinds, false);
472        <Settings::Form as FormAxis>::kinds(kinds).finish()
473    }
474
475    fn id(&self) -> &<Row as InboxTable>::Id {
476        InboxTable::id(self)
477    }
478
479    fn take_headers(&mut self) -> HeaderMap {
480        <Settings::Headers as HeadersAxis<Row>>::take(self)
481    }
482
483    fn unfit_header(headers: &HeaderMap) -> Option<&str> {
484        <Settings::Headers as HeadersAxis<Row>>::unfit(headers)
485    }
486
487    fn partition_key(&self) -> Option<&[u8]> {
488        <Settings::Key as KeyAxis<Row>>::key(self)
489    }
490
491    fn attempt(&self) -> Option<u64> {
492        <Settings::Attempt as AttemptAxis<DB, Row>>::attempt(self)
493    }
494
495    fn read_attempt(row: &DB::Row, _: &'static Queue) -> Option<u64> {
496        let column = Row::TABLE
497            .spec()
498            .column(Role::Attempt)
499            .map(|column| column.name());
500        <Settings::Attempt as AttemptAxis<DB, Row>>::read(row, column)
501    }
502
503    fn bind(
504        param: Param,
505        arguments: &mut DB::Arguments,
506        values: &Values<'_, DB, Self>,
507    ) -> Result<bool, Error> {
508        Ok(match (param, values.event) {
509            (Param::Id, _) => match values.id {
510                Some(id) => {
511                    engine::put::<DB, _>(arguments, id)?;
512                    true
513                }
514                None => false,
515            },
516            (Param::Group, _) => {
517                DB::bind_str(arguments, values.queue.name)?;
518                true
519            }
520            (Param::Limit, _) => {
521                DB::bind_i64(arguments, values.limit)?;
522                true
523            }
524            (Param::Destination, Event::DeadLetter) => {
525                DB::bind_str(arguments, values.destination)?;
526                true
527            }
528            (Param::Delay, Event::RetryAfter) => {
529                DB::bind_i64(arguments, engine::micros(values.delay))?;
530                true
531            }
532            (Param::Key, _) => match values.key {
533                Some(key) => {
534                    DB::bind_str(arguments, key)?;
535                    true
536                }
537                None => false,
538            },
539            (Param::Now, Event::Claim | Event::Take) => {
540                <Settings::RetryAfter as TimeAxis<DB>>::now::<Source<Settings>, Self>(
541                    arguments, values,
542                )?
543            }
544            (Param::RetryAfter, Event::RetryAfter) => {
545                <Settings::RetryAfter as TimeAxis<DB>>::later::<Source<Settings>, Self>(
546                    arguments, values,
547                )?
548            }
549            (Param::Now, Event::Ack | Event::Discard) => {
550                <Settings::ProcessedAt as TimeAxis<DB>>::now::<Source<Settings>, Self>(
551                    arguments, values,
552                )?
553            }
554            (Param::LeaseNow | Param::Lease | Param::Held, _) => {
555                <Settings::Form as FormBind<DB>>::bind(param, arguments, values)?
556            }
557            (Param::Ids, Event::Fetch) => {
558                <Own<Settings> as ClaimAxes<DB, Row>>::bind_ids(arguments, values)?
559            }
560            _ => false,
561        })
562    }
563
564    fn lease(queue: &'static Queue, now: Now) -> Result<Leasing<Token<Settings>>, Error> {
565        <Settings::Form as FormAxis>::lease::<Source<Settings>>(queue, now)
566    }
567
568    fn claim<'a>(
569        conn: &'a mut DB::Connection,
570        cx: &'a Claiming,
571        lease: Option<&'a Leasing<Token<Settings>>>,
572        ids: &'a mut Self::Ids,
573        out: &'a mut Vec<Claimed<Self>>,
574    ) -> impl Future<Output = Result<(), Error>> + Send + 'a {
575        <Own<Settings> as ClaimAxes<DB, Row>>::claim(conn, cx, lease, ids, out)
576    }
577
578    fn ack<'a>(
579        conn: &'a mut DB::Connection,
580        cx: &'a Settling,
581        id: &'a <Row as InboxTable>::Id,
582        held: Option<&'a Token<Settings>>,
583    ) -> impl Future<Output = Result<Settled, Error>> + Send + 'a {
584        <Settings::OwnAck as AckAxis<DB, Row>>::ack(conn, cx, id, held)
585    }
586
587    fn retry<'a>(
588        conn: &'a mut DB::Connection,
589        cx: &'a Settling,
590        id: &'a <Row as InboxTable>::Id,
591        held: Option<&'a Token<Settings>>,
592    ) -> impl Future<Output = Result<Settled, Error>> + Send + 'a {
593        <Settings::OwnRetry as RetryAxis<DB, Row>>::retry(conn, cx, id, held)
594    }
595
596    fn retry_after<'a>(
597        conn: &'a mut DB::Connection,
598        cx: &'a Settling,
599        id: &'a <Row as InboxTable>::Id,
600        held: Option<&'a Token<Settings>>,
601        delay: Duration,
602    ) -> impl Future<Output = Result<Settled, Error>> + Send + 'a {
603        <Settings::OwnRetryAfter as RetryAfterAxis<DB, Row>>::retry_after(conn, cx, id, held, delay)
604    }
605
606    fn discard<'a>(
607        conn: &'a mut DB::Connection,
608        cx: &'a Settling,
609        id: &'a <Row as InboxTable>::Id,
610        held: Option<&'a Token<Settings>>,
611    ) -> impl Future<Output = Result<Settled, Error>> + Send + 'a {
612        <Settings::OwnDiscard as DiscardAxis<DB, Row>>::discard(conn, cx, id, held)
613    }
614
615    fn dead_letter<'a>(
616        conn: &'a mut DB::Connection,
617        cx: &'a Settling,
618        id: &'a <Row as InboxTable>::Id,
619        held: Option<&'a Token<Settings>>,
620        destination: &'a str,
621    ) -> impl Future<Output = Result<Settled, Error>> + Send + 'a {
622        <Settings::OwnDeadLetter as DeadLetterAxis<DB, Row>>::dead_letter(
623            conn,
624            cx,
625            id,
626            held,
627            destination,
628        )
629    }
630
631    fn extend<'a>(
632        conn: &'a mut DB::Connection,
633        cx: &'a Settling,
634        id: &'a <Row as InboxTable>::Id,
635        held: &'a Token<Settings>,
636        until: &'a Token<Settings>,
637    ) -> impl Future<Output = Result<Settled, Error>> + Send + 'a {
638        <Settings::OwnExtend as ExtendAxis<DB, Row, Token<Settings>>>::extend(
639            conn, cx, id, held, until,
640        )
641    }
642
643    fn lock<'a>(
644        conn: &'a mut DB::Connection,
645        cx: &'a Claiming,
646        key: &'a str,
647    ) -> impl Future<Output = Result<bool, Error>> + Send + 'a {
648        <Settings::OwnLock as LockAxis<DB, Row>>::lock(conn, cx, key)
649    }
650
651    fn unlock<'a>(
652        conn: &'a mut DB::Connection,
653        cx: &'a Settling,
654        key: &'a str,
655    ) -> impl Future<Output = Result<bool, Error>> + Send + 'a {
656        <Settings::OwnUnlock as UnlockAxis<DB, Row>>::unlock(conn, cx, key)
657    }
658
659    fn take<'a>(
660        conn: &'a mut DB::Connection,
661        cx: &'a Claiming,
662        id: &'a <Row as InboxTable>::Id,
663        out: &'a mut Vec<Claimed<Self>>,
664    ) -> impl Future<Output = Result<bool, Error>> + Send + 'a {
665        <Own<Settings> as ClaimAxes<DB, Row>>::take(conn, cx, id, out)
666    }
667
668    fn candidates<'a>(
669        conn: &'a mut DB::Connection,
670        cx: &'a Claiming,
671        out: &'a mut Candidates<<Row as InboxTable>::Id>,
672    ) -> impl Future<Output = Result<(), Error>> + Send + 'a {
673        advisory::candidates::<DB, Self>(conn, cx, out)
674    }
675}
ruststream_sqlx/inbox/manual.rs

ruststream_sqlx/inbox/
manual.rs
