ruststream_sqlx/outbox/
wrap.rs

1//! A publisher with the publish middleware's tracking, for publishes outside the handlers.
2
3use std::fmt;
4use std::sync::Arc;
5
6use ruststream::{OutgoingFor, OutgoingMessage, Publisher};
7use sqlx::Database;
8
9use super::OUTBOX_ID_HEADER;
10use super::error::TrackedPublishError;
11use super::registry::RecordList;
12use super::switch::enabled;
13use crate::outbox::store::Store;
14
15/// A publisher that records what it publishes under a registered name, from
16/// [`wrap`](super::Outbox::wrap).
17///
18/// It publishes as the publisher it wraps does, with the same payload form and options. A message
19/// under a registered name is recorded first and carries [`OUTBOX_ID_HEADER`]; one under any
20/// other name goes out as it is.
21///
22/// # Examples
23///
24/// ```
25/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
26/// # mod demo {
27/// # use ruststream::OutgoingMessage;
28/// # use ruststream::memory::prelude::*;
29/// # use ruststream_sqlx::{Outbox, outbox};
30/// # use serde::{Deserialize, Serialize};
31/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
32/// # #[derive(Outbox, sqlx::FromRow)]
33/// # #[outbox(table = "outbox")]
34/// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
35/// # impl outbox::Publish<Postgres> for OrderOutbox {
36/// #     async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
37/// #         sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
38/// #             .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
39/// #     }
40/// # }
41/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
42/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
43/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
44/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
45/// pub async fn serve(pool: PgPool) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
46///     let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
47///     let broker = MemoryBroker::new().bindable();
48///     let egress = broker.bind(Publish);
49///     let running = RustStream::new(AppInfo::new("orders", "0.1.0"))
50///         .with_broker(broker, |b| {
51///             b.include(fulfil);
52///         })
53///         .start()
54///         .await?;
55///
56///     let publisher = tracking.wrap(running.publisher(egress).await?);
57///     publisher.message(&OrderPlaced { id: 7 }).publish().await?;
58///
59///     running.shutdown().await?;
60///     Ok(())
61/// }
62/// # }
63/// # fn main() {}
64/// ```
65pub struct TrackedPublisher<Live, DB: Database, Records> {
66    live: Live,
67    store: Arc<Store<DB>>,
68    records: Records,
69}
70
71impl<Live, DB: Database, Records> TrackedPublisher<Live, DB, Records> {
72    pub(super) const fn new(live: Live, store: Arc<Store<DB>>, records: Records) -> Self {
73        Self {
74            live,
75            store,
76            records,
77        }
78    }
79}
80
81impl<Live: Clone, DB: Database, Records: Copy> Clone for TrackedPublisher<Live, DB, Records> {
82    fn clone(&self) -> Self {
83        Self {
84            live: self.live.clone(),
85            store: Arc::clone(&self.store),
86            records: self.records,
87        }
88    }
89}
90
91impl<Live: fmt::Debug, DB: Database, Records: fmt::Debug> fmt::Debug
92    for TrackedPublisher<Live, DB, Records>
93{
94    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
95        f.debug_struct("TrackedPublisher")
96            .field("live", &self.live)
97            .field("records", &self.records)
98            .finish_non_exhaustive()
99    }
100}
101
102impl<Live, DB, Records> Publisher for TrackedPublisher<Live, DB, Records>
103where
104    Live: Publisher,
105    DB: Database,
106    Records: RecordList<DB>,
107{
108    type Payload = Live::Payload;
109    type Error = TrackedPublishError<Live::Error>;
110    type Options = Live::Options;
111
112    async fn publish(
113        &self,
114        msg: OutgoingFor<'_, Self::Payload>,
115        options: Option<&Self::Options>,
116    ) -> Result<(), Self::Error> {
117        let msg = if enabled() && self.records.contains(msg.name()) {
118            let (name, payload, headers) = msg.into_parts();
119            // The record reads a view that borrows the payload in whatever form the publisher
120            // takes it, so the payload is neither copied nor converted.
121            let view = OutgoingMessage::with_payload(name, payload.as_ref()).with_headers(headers);
122            let recorded = self.records.record(&self.store, &view).await;
123            let (_, _, mut headers) = view.into_parts();
124            if let Some(id) = recorded {
125                headers.insert(OUTBOX_ID_HEADER, id.map_err(TrackedPublishError::Outbox)?);
126            }
127            OutgoingMessage::with_payload(name, payload).with_headers(headers)
128        } else {
129            msg
130        };
131        self.live
132            .publish(msg, options)
133            .await
134            .map_err(TrackedPublishError::Publish)
135    }
136}
ruststream_sqlx/outbox/wrap.rs

ruststream_sqlx/outbox/
wrap.rs
