ruststream_sqlx/outbox/registry/
tracked.rs

1//! Names tracked as types: a name tracked twice stops the build.
2
3use std::future::Future;
4use std::marker::PhantomData;
5
6use ruststream::runtime::{Context, Handler, HandlerOutcome};
7use ruststream::{Bytes, OutgoingMessage, Publisher};
8use sqlx::Database;
9
10use super::list::{Nil, RecordList, RecordNames, Registered};
11use crate::outbox::error::OutboxError;
12use crate::outbox::store::Store;
13
14/// A name the outbox tracks, as a type: [`Outbox::track`](super::Outbox::track) refuses a name tracked twice while the
15/// service compiles.
16///
17/// # Examples
18///
19/// ```
20/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
21/// # mod demo {
22/// # use ruststream::OutgoingMessage;
23/// # use ruststream::memory::prelude::*;
24/// # use ruststream_sqlx::{Outbox, outbox};
25/// # use serde::{Deserialize, Serialize};
26/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
27/// # #[derive(Outbox, sqlx::FromRow)]
28/// # #[outbox(table = "outbox")]
29/// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
30/// # impl outbox::Publish<Postgres> for OrderOutbox {
31/// #     async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
32/// #         sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
33/// #             .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
34/// #     }
35/// # }
36/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
37/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
38/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
39/// use ruststream_sqlx::outbox::TrackedName;
40///
41/// /// The name `OrderPlaced` is published under, as a type.
42/// pub struct Orders;
43///
44/// impl TrackedName for Orders {
45///     const NAME: &'static str = "orders";
46/// }
47///
48/// #[derive(Serialize, Deserialize, Outgoing)]
49/// #[outgoing(name = "orders")]
50/// pub struct OrderPlaced {
51///     id: u64,
52/// }
53///
54/// pub fn app(pool: PgPool) -> impl App {
55///     let tracking = Outbox::new(pool).track::<OrderOutbox, Orders>();
56///     RustStream::new(AppInfo::new("orders", "0.1.0"))
57///         .layer(tracking.layer())
58///         .publish_layer(tracking.publish_layer())
59///         .with_broker(MemoryBroker::new(), |b| {
60///             b.include(place).out_reply(Publish);
61///             b.include(fulfil);
62///         })
63/// }
64/// # }
65/// # fn main() {}
66/// ```
67#[diagnostic::on_unimplemented(
68    message = "`{Self}` is not a name the outbox tracks",
69    label = "not a `TrackedName`",
70    note = "implement `TrackedName` for `{Self}` with the name as `NAME`, or register the name as \
71            a string with `register`"
72)]
73pub trait TrackedName: 'static {
74    /// The name.
75    const NAME: &'static str;
76}
77
78/// A name tracked by type, kept in the registry's type for the compile-time check; it holds no
79/// record and forwards every call. Machinery behind [`Outbox::track`](super::Outbox::track).
80#[derive(Debug)]
81pub struct Checked<Name, Rest>(pub(super) Rest, pub(super) PhantomData<fn() -> Name>);
82
83impl<Name, Rest: Clone> Clone for Checked<Name, Rest> {
84    fn clone(&self) -> Self {
85        Self(self.0.clone(), PhantomData)
86    }
87}
88
89impl<Name, Rest: Copy> Copy for Checked<Name, Rest> {}
90
91/// Whether a list of registrations lacks `Name` among the names tracked by type. Machinery.
92#[doc(hidden)]
93pub trait Lacks<Name: TrackedName> {
94    /// `true` when no node tracks `Name::NAME` by type.
95    const LACKS: bool;
96}
97
98impl<Name: TrackedName> Lacks<Name> for Nil {
99    const LACKS: bool = true;
100}
101
102impl<Name: TrackedName, Record, Rest: Lacks<Name>> Lacks<Name> for Registered<Record, Rest> {
103    const LACKS: bool = <Rest as Lacks<Name>>::LACKS;
104}
105
106impl<Name: TrackedName, Earlier: TrackedName, Rest: Lacks<Name>> Lacks<Name>
107    for Checked<Earlier, Rest>
108{
109    const LACKS: bool = !same(Earlier::NAME, Name::NAME) && <Rest as Lacks<Name>>::LACKS;
110}
111
112/// Whether two names are one, while the service compiles.
113const fn same(one: &str, other: &str) -> bool {
114    let (one, other) = (one.as_bytes(), other.as_bytes());
115    if one.len() != other.len() {
116        return false;
117    }
118    let mut index = 0;
119    while index < one.len() {
120        if one[index] != other[index] {
121            return false;
122        }
123        index += 1;
124    }
125    true
126}
127
128impl<Name: 'static, Rest: RecordNames> RecordNames for Checked<Name, Rest> {
129    #[inline]
130    fn contains(&self, name: &str) -> bool {
131        self.0.contains(name)
132    }
133}
134
135impl<DB: Database, Name: 'static, Rest: RecordList<DB>> RecordList<DB> for Checked<Name, Rest> {
136    fn record<'a>(
137        &'a self,
138        pool: &'a Store<DB>,
139        msg: &'a OutgoingMessage<'_>,
140    ) -> impl Future<Output = Option<Result<Bytes, OutboxError>>> + Send + 'a {
141        self.0.record(pool, msg)
142    }
143
144    fn deliver<'a, M, C, S, H>(
145        &'a self,
146        pool: &'a Store<DB>,
147        handler: &'a H,
148        msg: &'a M,
149        ctx: &'a mut Context<'_, C, S>,
150    ) -> impl Future<Output = HandlerOutcome> + Send + 'a
151    where
152        M: Sync,
153        C: Send,
154        S: Send + Sync,
155        H: Handler<M, C, S>,
156    {
157        self.0.deliver(pool, handler, msg, ctx)
158    }
159
160    fn republish<'a, Live: Publisher>(
161        &'a self,
162        pool: &'a Store<DB>,
163        publisher: &'a Live,
164        only: Option<&'a [&'static str]>,
165    ) -> impl Future<Output = Result<(), OutboxError>> + Send + 'a {
166        self.0.republish(pool, publisher, only)
167    }
168}
169
170#[cfg(test)]
171mod tests {
172    use super::{Checked, Lacks, Nil, PhantomData, RecordNames, Registered, TrackedName};
173    use crate::outbox::registry::list::tests::node;
174
175    struct Orders;
176
177    impl TrackedName for Orders {
178        const NAME: &'static str = "orders";
179    }
180
181    struct Order;
182
183    impl TrackedName for Order {
184        const NAME: &'static str = "order";
185    }
186
187    struct Refunds;
188
189    impl TrackedName for Refunds {
190        const NAME: &'static str = "refunds";
191    }
192
193    type Tracked = Registered<(), Checked<Refunds, Registered<(), Checked<Orders, Nil>>>>;
194
195    #[test]
196    fn a_name_tracked_by_type_is_found_by_its_text_alone() {
197        const {
198            assert!(!<Tracked as Lacks<Orders>>::LACKS);
199            assert!(!<Tracked as Lacks<Refunds>>::LACKS);
200            assert!(<Tracked as Lacks<Order>>::LACKS);
201            assert!(<Nil as Lacks<Orders>>::LACKS);
202        }
203        let names = node(
204            "refunds",
205            Checked(node("orders", Nil), PhantomData::<fn() -> Refunds>),
206        );
207        assert!(names.contains("orders"));
208        assert!(names.contains("refunds"));
209        assert!(!names.contains("order"));
210    }
211}
ruststream_sqlx/outbox/registry/tracked.rs

ruststream_sqlx/outbox/registry/
tracked.rs
