ruststream_sqlx/outbox/
publish.rs

1//! The publish middleware: a message published under a registered name is recorded before it is
2//! sent, and carries its record's id.
3
4use std::any::type_name;
5use std::error::Error as StdError;
6use std::fmt::{self, Display};
7use std::io::Write;
8use std::mem;
9use std::str::{self, FromStr};
10use std::sync::Arc;
11
12use ruststream::runtime::{Outgoing, PublishLayer, PublishNext, PublishPipeline};
13use ruststream::{Bytes, HeaderMap, OutgoingMessage, Publisher};
14use sqlx::Database;
15
16use super::OUTBOX_ID_HEADER;
17use super::error::OutboxError;
18use super::events::Publish;
19use super::registry::RecordList;
20use super::switch::enabled;
21use crate::outbox::store::Store;
22
23/// The publish middleware of an [`Outbox`](super::Outbox), from
24/// [`publish_layer`](super::Outbox::publish_layer).
25///
26/// A message under a registered name is recorded through its record type's `Publish`, then sent
27/// with [`OUTBOX_ID_HEADER`]; a record that fails fails the publish, and the message is not sent.
28/// A message under any other name is sent as it is.
29pub struct TrackingPublishLayer<DB: Database, Records> {
30    store: Arc<Store<DB>>,
31    records: Records,
32}
33
34impl<DB: Database, Records> TrackingPublishLayer<DB, Records> {
35    pub(super) const fn new(store: Arc<Store<DB>>, records: Records) -> Self {
36        Self { store, records }
37    }
38}
39
40impl<DB: Database, Records: Copy> Clone for TrackingPublishLayer<DB, Records> {
41    fn clone(&self) -> Self {
42        Self {
43            store: Arc::clone(&self.store),
44            records: self.records,
45        }
46    }
47}
48
49impl<DB: Database, Records: fmt::Debug> fmt::Debug for TrackingPublishLayer<DB, Records> {
50    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
51        f.debug_struct("TrackingPublishLayer")
52            .field("records", &self.records)
53            .finish_non_exhaustive()
54    }
55}
56
57impl<DB: Database, Records: RecordList<DB>> PublishLayer for TrackingPublishLayer<DB, Records> {
58    async fn on_publish<'a, N: PublishPipeline, P: Publisher>(
59        &'a self,
60        out: &'a mut Outgoing<'a>,
61        next: PublishNext<'a, N, P>,
62    ) -> Result<(), Box<dyn StdError + Send + Sync>> {
63        if enabled() && self.records.contains(out.name()) {
64            // The map moves out and back: the record reads it through an `OutgoingMessage`
65            // that borrows the name and the payload, so nothing is copied.
66            let headers = mem::take(out.headers_mut());
67            let msg =
68                OutgoingMessage::with_payload(out.name(), out.payload()).with_headers(headers);
69            let recorded = self.records.record(&self.store, &msg).await;
70            let (_, _, headers) = msg.into_parts();
71            *out.headers_mut() = headers;
72            if let Some(id) = recorded {
73                out.headers_mut().insert(OUTBOX_ID_HEADER, id?);
74            }
75        }
76        next.run(out).await
77    }
78}
79
80/// Records `msg` with `Record`, and returns the id header's value.
81pub(super) async fn record<DB: Database, Record: Publish<DB>>(
82    name: &'static str,
83    store: &Store<DB>,
84    msg: &OutgoingMessage<'_>,
85) -> Result<Bytes, OutboxError> {
86    let pool = store.get().ok_or(OutboxError::NoPool { name })?;
87    let failed = |source| OutboxError::Record {
88        name,
89        record: type_name::<Record>(),
90        source,
91    };
92    let mut conn = store.acquire(pool).await.map_err(failed)?;
93    let id = Record::publish(&mut conn, msg).await.map_err(failed)?;
94    Ok(id_value(&id))
95}
96
97/// The longest id text [`id_value`] formats on the stack: a UUID's 36 characters fit.
98const ID_ON_STACK: usize = 64;
99
100/// The id header's value: the id's `Display` text in a buffer of its own length.
101///
102/// One allocation. `to_string` would take two: its `String` reserves more than an integer's
103/// digits, and a `Bytes` made from a vector with spare capacity allocates a block to share it.
104pub(super) fn id_value(id: &impl Display) -> Bytes {
105    let mut text = [0_u8; ID_ON_STACK];
106    let mut rest = &mut text[..];
107    if write!(rest, "{id}").is_ok() {
108        let len = ID_ON_STACK - rest.len();
109        return Bytes::copy_from_slice(&text[..len]);
110    }
111    Bytes::copy_from_slice(id.to_string().as_bytes())
112}
113
114/// The id a delivery carries in [`OUTBOX_ID_HEADER`]: `None` without the header, the header's
115/// text when it does not parse as an `Id`.
116pub(super) fn carried_id<Id: FromStr>(headers: &HeaderMap) -> Option<Result<Id, String>> {
117    let value = headers.get(OUTBOX_ID_HEADER)?;
118    Some(
119        str::from_utf8(value)
120            .ok()
121            .and_then(|text| text.parse().ok())
122            .ok_or_else(|| String::from_utf8_lossy(value).into_owned()),
123    )
124}
125
126#[cfg(test)]
127mod tests {
128    use ruststream::HeaderMap;
129
130    use super::{ID_ON_STACK, OUTBOX_ID_HEADER, carried_id, id_value};
131
132    #[test]
133    fn an_id_survives_the_header() {
134        let mut headers = HeaderMap::new();
135        headers.insert(OUTBOX_ID_HEADER, 42_i64.to_string());
136        assert_eq!(carried_id::<i64>(&headers), Some(Ok(42)));
137        headers.insert(OUTBOX_ID_HEADER, "7f3a".to_owned());
138        assert_eq!(carried_id::<String>(&headers), Some(Ok("7f3a".to_owned())));
139    }
140
141    #[test]
142    fn an_id_value_holds_the_ids_text_whatever_its_length() {
143        assert_eq!(&id_value(&42_i64)[..], b"42");
144        assert_eq!(&id_value(&i64::MIN)[..], i64::MIN.to_string().as_bytes());
145        let long = "7".repeat(ID_ON_STACK + 1);
146        assert_eq!(&id_value(&long)[..], long.as_bytes());
147    }
148
149    #[test]
150    fn a_missing_or_foreign_header_is_told_apart() {
151        let mut headers = HeaderMap::new();
152        assert_eq!(carried_id::<i64>(&headers), None);
153        headers.insert(OUTBOX_ID_HEADER, "seven");
154        assert_eq!(carried_id::<i64>(&headers), Some(Err("seven".to_owned())));
155        headers.insert(OUTBOX_ID_HEADER, &b"\xff7"[..]);
156        assert_eq!(
157            carried_id::<i64>(&headers),
158            Some(Err("\u{fffd}7".to_owned()))
159        );
160    }
161}
ruststream_sqlx/outbox/publish.rs

ruststream_sqlx/outbox/
publish.rs
