ruststream_sqlx/outbox/
republish.rs

1//! The startup republish: every unprocessed record of a registered name is published again, with
2//! its id.
3
4use std::any::type_name;
5use std::fmt;
6use std::future::Future;
7use std::pin::Pin;
8use std::task::{Context, Poll};
9
10use ruststream::{OutgoingMessage, PayloadForm, Publisher};
11use sqlx::Database;
12
13use super::OUTBOX_ID_HEADER;
14use super::database::Defaults;
15use super::error::OutboxError;
16use super::events::Tracked;
17use super::publish::id_value;
18use super::switch::enabled;
19use crate::outbox::store::Store;
20
21/// The body of the startup republish, which [`republish`](super::Outbox::republish) hands to
22/// `after_startup`: it resolves once every unprocessed record went out again, and fails startup
23/// with the first record that did not.
24#[must_use = "a republish does nothing until it is awaited"]
25pub struct Republishing(
26    // Why boxed: the hook's return type has to be named, and stable Rust cannot name an async
27    // block's type in a closure's return. One allocation and one dynamic poll, once per startup.
28    Pin<Box<dyn Future<Output = Result<(), OutboxError>> + Send>>,
29);
30
31impl Republishing {
32    pub(super) fn new(
33        body: impl Future<Output = Result<(), OutboxError>> + Send + 'static,
34    ) -> Self {
35        Self(Box::pin(async move {
36            if enabled() { body.await } else { Ok(()) }
37        }))
38    }
39}
40
41impl Future for Republishing {
42    type Output = Result<(), OutboxError>;
43
44    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
45        self.get_mut().0.as_mut().poll(cx)
46    }
47}
48
49impl fmt::Debug for Republishing {
50    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
51        f.debug_struct("Republishing").finish_non_exhaustive()
52    }
53}
54
55/// Publishes every unprocessed record `Record` of `name` through `publisher`.
56pub(super) async fn recover_and_publish<DB, Record, Live>(
57    name: &'static str,
58    defaults: &Defaults,
59    pool: &Store<DB>,
60    publisher: &Live,
61) -> Result<(), OutboxError>
62where
63    DB: Database,
64    Record: Tracked<DB>,
65    Live: Publisher,
66{
67    let pool = pool.get().ok_or(OutboxError::NoPool { name })?;
68    let recovered = async {
69        let mut conn = pool.acquire().await?;
70        Record::recover_records(&mut conn, name, defaults).await
71    }
72    .await
73    .map_err(|source| OutboxError::Recover {
74        name,
75        record: type_name::<Record>(),
76        source,
77    })?;
78    for mut record in recovered {
79        let mut headers = record.take_headers();
80        headers.insert(OUTBOX_ID_HEADER, id_value(record.id()));
81        let payload = <Live::Payload as PayloadForm>::Form::from(record.payload());
82        let msg = OutgoingMessage::with_payload(record.name(), payload).with_headers(headers);
83        if let Err(source) = publisher.publish(msg, None).await {
84            return Err(OutboxError::Republish {
85                name,
86                record: type_name::<Record>(),
87                id: record.id().to_string(),
88                source: Box::new(source),
89            });
90        }
91    }
92    Ok(())
93}
ruststream_sqlx/outbox/republish.rs

ruststream_sqlx/outbox/
republish.rs
