ruststream_sqlx/inbox/
columns.rs

1//! How the columns of a queue row reach a delivery: its key, its attempt.
2
3/// The type of a field playing `partition_key`: the bytes the delivery's key lends.
4///
5/// # Examples
6///
7/// ```
8/// # #[cfg(feature = "postgres")]
9/// # mod demo {
10/// use ruststream_sqlx::prelude::*;
11/// use serde::Deserialize;
12/// use sqlx::PgPool;
13///
14/// #[derive(Inbox, sqlx::FromRow)]
15/// #[inbox(table = "order_jobs")]
16/// pub struct OrderJob {
17///     #[field(id, generated)]
18///     id: i64,
19///     // A `String` lends its bytes as the delivery's key; an empty `Option` gives none.
20///     #[field(partition_key)]
21///     customer: Option<String>,
22///     #[field(payload)]
23///     payload: Vec<u8>,
24/// }
25///
26/// #[derive(Deserialize)]
27/// struct Order {
28///     id: u64,
29/// }
30///
31/// #[subscriber(InboxQueue::<OrderJob>::new("orders"))]
32/// async fn fulfil(order: &Order) -> HandlerOutcome {
33///     tracing::info!(order.id, "fulfilling");
34///     HandlerOutcome::ack()
35/// }
36///
37/// pub fn app(pool: PgPool) -> RustStream {
38///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
39///         // Four workers, and the orders of one customer keep their order on one of them.
40///         b.include(fulfil.workers_by_key(nonzero!(4)));
41///     })
42/// }
43/// # }
44/// # fn main() {}
45/// ```
46pub trait KeyColumn {
47    /// The key's bytes, or `None` when the row carries no key.
48    fn key(&self) -> Option<&[u8]>;
49}
50
51macro_rules! keys {
52    ($($ty:ty => $bytes:ident),*) => {$(
53        impl KeyColumn for $ty {
54            fn key(&self) -> Option<&[u8]> {
55                Some(self.$bytes())
56            }
57        }
58    )*};
59}
60
61keys!(String => as_bytes, Box<str> => as_bytes, Vec<u8> => as_slice, Box<[u8]> => as_ref);
62
63impl<T: KeyColumn> KeyColumn for Option<T> {
64    fn key(&self) -> Option<&[u8]> {
65        self.as_ref().and_then(KeyColumn::key)
66    }
67}
68
69/// The type of a field playing `attempt`: the delivery's redelivery count.
70///
71/// The first delivery of a row reads 1, so a table's `attempt` column starts at 1. A negative
72/// value reads as 0.
73///
74/// # Examples
75///
76/// ```
77/// # #[cfg(feature = "postgres")]
78/// # mod demo {
79/// use ruststream_sqlx::prelude::*;
80/// use serde::Deserialize;
81/// use sqlx::PgPool;
82///
83/// #[derive(Inbox, sqlx::FromRow)]
84/// #[inbox(table = "charge_jobs")]
85/// pub struct ChargeJob {
86///     #[field(id, generated)]
87///     id: i64,
88///     // sqlx has no unsigned integers on Postgres, so the column is a `SMALLINT DEFAULT 1`.
89///     #[field(attempt, generated)]
90///     attempt: i16,
91///     #[field(payload)]
92///     payload: Vec<u8>,
93/// }
94///
95/// #[derive(Deserialize)]
96/// struct Charge {
97///     order: u64,
98/// }
99///
100/// # async fn try_charge(_: &Charge) -> bool { true }
101/// #[subscriber(InboxQueue::<ChargeJob>::new("charges"))]
102/// async fn charge(request: &Charge, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
103///     tracing::info!(request.order, ?attempt, "charging");
104///     if try_charge(request).await { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
105/// }
106///
107/// pub fn app(pool: PgPool) -> RustStream {
108///     RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
109///         // The attempt counts the deliveries: the fifth failure moves the row into
110///         // `failed_charge_jobs`, a table with the same columns.
111///         b.include(charge).max_attempts(nonzero!(5u32)).dead_letter("failed_charge_jobs");
112///     })
113/// }
114/// # }
115/// # fn main() {}
116/// ```
117pub trait AttemptColumn {
118    /// The attempt, counting the first delivery as 1.
119    fn attempt(&self) -> u64;
120}
121
122macro_rules! attempts {
123    ($($ty:ty),*) => {$(
124        impl AttemptColumn for $ty {
125            fn attempt(&self) -> u64 {
126                u64::try_from(*self).unwrap_or(0)
127            }
128        }
129    )*};
130}
131
132attempts!(i16, i32, i64, u16, u32, u64);
133
134#[cfg(test)]
135mod tests {
136    use super::{AttemptColumn, KeyColumn};
137
138    #[test]
139    fn keys_lend_their_bytes() {
140        assert_eq!(Box::<str>::from("k").key(), Some(b"k".as_slice()));
141        assert_eq!(Some(vec![9_u8]).key(), Some([9].as_slice()));
142    }
143
144    #[test]
145    fn attempts_never_read_below_zero() {
146        assert_eq!(i64::MIN.attempt(), 0);
147        assert_eq!(u64::MAX.attempt(), u64::MAX);
148    }
149}
ruststream_sqlx/inbox/columns.rs

ruststream_sqlx/inbox/
columns.rs
