ruststream_sqlx/
outbox.rs

1//! The transactional outbox: what a service publishes is recorded in its own table until a
2//! consumer has processed it, and what was not processed is published again at startup.
3//!
4//! A struct deriving [`Outbox`](derive@crate::Outbox), or implementing [`OutboxTable`] by hand,
5//! describes the table; `outbox!`, [`Outbox::register`] or [`Outbox::track`] registers it under the
6//! names it tracks, and the registry hands out the two middlewares and the republish. The events
7//! below are what the record's statements do; a service implements the ones it names as its own
8//! (`#[outbox(custom(..))]`, or [`OutboxSpec::own`]), and always [`Publish`], which has no
9//! default.
10//!
11//! The crate overview's [transactional outbox](crate#the-transactional-outbox) section shows a
12//! whole service and states the guarantee, the test switch and the costs.
13
14mod database;
15mod dispatch;
16mod error;
17mod events;
18mod layer;
19mod publish;
20mod registry;
21mod republish;
22mod row;
23pub mod spec;
24mod store;
25mod switch;
26mod wrap;
27
28pub use database::OutboxDatabase;
29#[doc(hidden)]
30pub use database::{Defaults, Statements};
31#[doc(hidden)]
32pub use dispatch::{AckBy, DiscardBy, FetchBy, HeadersOf, RecoverBy, RetryBy, Slot};
33pub use error::{OutboxError, PoolAlreadySet, TrackedPublishError};
34pub use events::{Ack, Discard, Fetch, Publish, Recover, Retry, Tracked};
35pub use layer::TrackingLayer;
36pub use publish::TrackingPublishLayer;
37pub use registry::{Checked, Nil, Outbox, Registered, TrackedName};
38#[doc(hidden)]
39pub use registry::{Lacks, RecordList, RecordNames};
40pub use republish::Republishing;
41pub use row::OutboxRow;
42pub use spec::{OutboxSpec, OutboxTable};
43#[doc(hidden)]
44pub use store::Store;
45pub use wrap::TrackedPublisher;
46
47/// The header a tracked message carries its record's id in: written with the id's `Display`,
48/// read back with its `FromStr`.
49///
50/// # Examples
51///
52/// ```
53/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
54/// # mod demo {
55/// # use ruststream::OutgoingMessage;
56/// # use ruststream::memory::prelude::*;
57/// # use ruststream_sqlx::{Outbox, outbox};
58/// # use serde::{Deserialize, Serialize};
59/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
60/// # #[derive(Outbox, sqlx::FromRow)]
61/// # #[outbox(table = "outbox")]
62/// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
63/// # impl outbox::Publish<Postgres> for OrderOutbox {
64/// #     async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
65/// #         sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
66/// #             .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
67/// #     }
68/// # }
69/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
70/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
71/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
72/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
73/// use ruststream::runtime::{BlanketLayer, Context, Handler};
74/// use ruststream_sqlx::outbox::OUTBOX_ID_HEADER;
75///
76/// /// A layer of the service's own that logs the record id of every tracked delivery.
77/// #[derive(Clone, Copy)]
78/// pub struct RecordIds;
79///
80/// pub struct Logged<H>(H);
81///
82/// impl BlanketLayer for RecordIds {
83///     fn apply<M, C, S, H>(&self, handler: H) -> impl Handler<M, C, S> + 'static
84///     where
85///         M: Send + Sync + 'static,
86///         C: Send + 'static,
87///         S: Send + Sync + 'static,
88///         H: Handler<M, C, S> + 'static,
89///     {
90///         Logged(handler)
91///     }
92/// }
93///
94/// impl<M: Sync, C: Send, S: Send + Sync, H: Handler<M, C, S>> Handler<M, C, S> for Logged<H> {
95///     async fn handle(&self, msg: &M, ctx: &mut Context<'_, C, S>) -> HandlerOutcome {
96///         if let Some(id) = ctx.headers().get(OUTBOX_ID_HEADER) {
97///             let id = String::from_utf8_lossy(id);
98///             tracing::info!(subscription = ctx.name(), %id, "a tracked delivery");
99///         }
100///         self.0.handle(msg, ctx).await
101///     }
102/// }
103///
104/// pub fn app(pool: PgPool) -> impl App {
105///     let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
106///     RustStream::new(AppInfo::new("orders", "0.1.0"))
107///         .layer(RecordIds)
108///         .layer(tracking.layer())
109///         .with_broker(MemoryBroker::new(), |b| {
110///             b.include(fulfil);
111///         })
112/// }
113/// # }
114/// # fn main() {}
115/// ```
116pub const OUTBOX_ID_HEADER: &str = "x-ruststream-outbox-id";
ruststream_sqlx/outbox.rs

ruststream_sqlx/
outbox.rs
