ruststream_sqlx/outbox/registry.rs
1//! The registry: which record type tracks which name, and the pool every handle it gives out
2//! shares.
3//!
4//! The registrations are a type-level list, `Registered<Record, Rest>` ending in [`Nil`], each
5//! node holding its name and its record's default statements, built when the record type was
6//! registered. A lookup walks the list comparing names, and every node is its own monomorphized
7//! code: no map, no `dyn`. A name tracked by type adds a [`Checked`] node, which holds nothing and
8//! forwards every call.
9
10use std::fmt;
11use std::marker::PhantomData;
12use std::sync::Arc;
13
14use ruststream::Publisher;
15use sqlx::{Database, Pool};
16
17use super::database::OutboxDatabase;
18use super::error::PoolAlreadySet;
19use super::layer::TrackingLayer;
20use super::publish::TrackingPublishLayer;
21use super::republish::Republishing;
22use super::spec::OutboxTable;
23use super::wrap::TrackedPublisher;
24use crate::outbox::store::Store;
25
26mod list;
27mod tracked;
28
29use list::defaults;
30pub use list::{Nil, RecordList, RecordNames, Registered};
31pub use tracked::{Checked, Lacks, TrackedName};
32
33/// The outbox of a service: the record type of each name it tracks, and the pool their records
34/// live in.
35///
36/// It hands out the subscription middleware ([`layer`](Self::layer)), the publish middleware
37/// ([`publish_layer`](Self::publish_layer)), the startup republish
38/// ([`republish`](Self::republish)) and [`wrap`](Self::wrap) for a publisher outside the
39/// handlers. Every handle shares one pool: a pool set later, with [`set_pool`](Self::set_pool),
40/// reaches the handles given out before.
41///
42/// A message whose name is not registered passes every handle after a comparison with each
43/// registered name, and nothing else. A tracked message reads the pool with one atomic load, takes
44/// a connection for its statements, and allocates the id header's value.
45///
46/// # Examples
47///
48/// ```
49/// # #[cfg(all(feature = "outbox", feature = "postgres"))]
50/// # mod demo {
51/// # use ruststream::OutgoingMessage;
52/// # use ruststream::memory::prelude::*;
53/// # use ruststream_sqlx::{Outbox, outbox};
54/// # use serde::{Deserialize, Serialize};
55/// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
56/// # #[derive(Outbox, sqlx::FromRow)]
57/// # #[outbox(table = "outbox")]
58/// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
59/// # impl outbox::Publish<Postgres> for OrderOutbox {
60/// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
61/// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
62/// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
63/// # }
64/// # }
65/// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
66/// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
67/// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
68/// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
69/// use std::convert::Infallible;
70/// use std::io;
71///
72/// /// What `#[ruststream::app]` runs: the pool is built in `on_startup`, inside the runtime.
73/// pub fn app() -> impl App {
74/// let tracking = Outbox::<Postgres>::deferred().register::<OrderOutbox>("orders");
75/// let registry = tracking.clone();
76/// RustStream::new(AppInfo::new("orders", "0.1.0"))
77/// .on_startup(async move |()| {
78/// let pool = PgPool::connect("postgres://localhost/orders")
79/// .await
80/// .map_err(io::Error::other)?;
81/// registry.set_pool(pool.clone()).map_err(io::Error::other)?;
82/// Ok::<_, io::Error>(pool)
83/// })
84/// .layer(tracking.layer())
85/// .publish_layer(tracking.publish_layer())
86/// .after_shutdown(async move |pool| {
87/// pool.close().await;
88/// Ok::<_, Infallible>(())
89/// })
90/// .with_broker(MemoryBroker::new(), |b| {
91/// b.include(place).out_reply(Publish);
92/// b.include(fulfil);
93/// b.after_startup(Publish, tracking.republish());
94/// })
95/// }
96/// # }
97/// # fn main() {}
98/// ```
99pub struct Outbox<DB: Database, Records = Nil> {
100 store: Arc<Store<DB>>,
101 records: Records,
102}
103
104impl<DB: Database, Records: Copy> Clone for Outbox<DB, Records> {
105 fn clone(&self) -> Self {
106 Self {
107 store: Arc::clone(&self.store),
108 records: self.records,
109 }
110 }
111}
112
113impl<DB: Database, Records: fmt::Debug> fmt::Debug for Outbox<DB, Records> {
114 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
115 f.debug_struct("Outbox")
116 .field("records", &self.records)
117 .field("pool_set", &self.store.get().is_some())
118 .finish()
119 }
120}
121
122impl<DB: Database> Outbox<DB, Nil> {
123 /// An outbox with no names yet, whose records live in `pool`.
124 ///
125 /// # Examples
126 ///
127 /// ```
128 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
129 /// # mod demo {
130 /// # use ruststream::OutgoingMessage;
131 /// # use ruststream::memory::prelude::*;
132 /// # use ruststream_sqlx::{Outbox, outbox};
133 /// # use serde::{Deserialize, Serialize};
134 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
135 /// # #[derive(Outbox, sqlx::FromRow)]
136 /// # #[outbox(table = "outbox")]
137 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
138 /// # impl outbox::Publish<Postgres> for OrderOutbox {
139 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
140 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
141 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
142 /// # }
143 /// # }
144 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
145 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
146 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
147 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
148 /// /// An app built inside a running Tokio runtime, where the pool exists already.
149 /// pub async fn run(pool: PgPool) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
150 /// let tracking = Outbox::new(pool).register::<OrderOutbox>("orders");
151 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
152 /// .layer(tracking.layer())
153 /// .publish_layer(tracking.publish_layer())
154 /// .with_broker(MemoryBroker::new(), |b| {
155 /// b.include(place).out_reply(Publish);
156 /// b.include(fulfil);
157 /// b.after_startup(Publish, tracking.republish());
158 /// })
159 /// .run()
160 /// .await?;
161 /// Ok(())
162 /// }
163 /// # }
164 /// # fn main() {}
165 /// ```
166 #[must_use]
167 pub fn new(pool: Pool<DB>) -> Self {
168 Self {
169 store: Arc::new(Store::from(pool)),
170 records: Nil,
171 }
172 }
173
174 /// An outbox with no names and no pool yet: the service gives it the pool with
175 /// [`set_pool`](Self::set_pool), usually in `on_startup`, where the pool is built.
176 ///
177 /// # Examples
178 ///
179 /// ```
180 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
181 /// # mod demo {
182 /// # use ruststream::OutgoingMessage;
183 /// # use ruststream::memory::prelude::*;
184 /// # use ruststream_sqlx::{Outbox, outbox};
185 /// # use serde::{Deserialize, Serialize};
186 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
187 /// # #[derive(Outbox, sqlx::FromRow)]
188 /// # #[outbox(table = "outbox")]
189 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
190 /// # impl outbox::Publish<Postgres> for OrderOutbox {
191 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
192 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
193 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
194 /// # }
195 /// # }
196 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
197 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
198 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
199 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
200 /// use std::io;
201 ///
202 /// pub fn app() -> impl App {
203 /// let tracking = Outbox::<Postgres>::deferred().register::<OrderOutbox>("orders");
204 /// let registry = tracking.clone();
205 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
206 /// .on_startup(async move |()| {
207 /// let pool = PgPool::connect("postgres://localhost/orders")
208 /// .await
209 /// .map_err(io::Error::other)?;
210 /// registry.set_pool(pool.clone()).map_err(io::Error::other)?;
211 /// Ok::<_, io::Error>(pool)
212 /// })
213 /// .publish_layer(tracking.publish_layer())
214 /// .with_broker(MemoryBroker::new(), |b| {
215 /// b.include(place).out_reply(Publish);
216 /// })
217 /// }
218 /// # }
219 /// # fn main() {}
220 /// ```
221 #[must_use]
222 pub fn deferred() -> Self {
223 Self {
224 store: Arc::new(Store::default()),
225 records: Nil,
226 }
227 }
228}
229
230impl<DB: Database, Records: RecordNames> Outbox<DB, Records> {
231 /// Tracks the messages published under `name` with the record type `Record`, a name the
232 /// service passes as a `&'static str` value; [`track`](Self::track) takes the name as a type
233 /// and refuses a repeated name while the service compiles, and here a repeated name panics
234 /// when the app is built. The record's default statements are built here, once.
235 ///
236 /// # Panics
237 ///
238 /// Panics when `name` is registered already (each name has one record type), and when
239 /// `Record`'s table does not fit a dialect of `DB` (an identifier longer than the database
240 /// takes).
241 ///
242 /// # Examples
243 ///
244 /// ```
245 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
246 /// # mod demo {
247 /// # use ruststream::OutgoingMessage;
248 /// # use ruststream::memory::prelude::*;
249 /// # use ruststream_sqlx::{Outbox, outbox};
250 /// # use serde::{Deserialize, Serialize};
251 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
252 /// # #[derive(Outbox, sqlx::FromRow)]
253 /// # #[outbox(table = "outbox")]
254 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
255 /// # impl outbox::Publish<Postgres> for OrderOutbox {
256 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
257 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
258 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
259 /// # }
260 /// # }
261 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
262 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
263 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
264 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
265 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "refunds")] pub struct RefundRequested { id: u64 }
266 /// # #[subscriber("cancellations", reply)] async fn cancel(cmd: &PlaceOrder) -> RefundRequested { RefundRequested { id: cmd.id } }
267 /// /// One record type tracks both names; each name republishes its own records.
268 /// pub fn app(pool: PgPool) -> impl App {
269 /// let tracking = Outbox::new(pool)
270 /// .register::<OrderOutbox>("orders")
271 /// .register::<OrderOutbox>("refunds");
272 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
273 /// .publish_layer(tracking.publish_layer())
274 /// .with_broker(MemoryBroker::new(), |b| {
275 /// b.include(place).out_reply(Publish);
276 /// b.include(cancel).out_reply(Publish);
277 /// b.after_startup(Publish, tracking.republish());
278 /// })
279 /// }
280 /// # }
281 /// # fn main() {}
282 /// ```
283 #[must_use]
284 #[track_caller]
285 pub fn register<Record: OutboxTable>(
286 self,
287 name: &'static str,
288 ) -> Outbox<DB, Registered<Record, Records>>
289 where
290 DB: OutboxDatabase,
291 // On the node rather than on `Record`: rustc then reports the unmet bound underneath (a
292 // missing `Publish`, an event of the record's own without its impl) with its own message.
293 Registered<Record, Records>: RecordList<DB>,
294 {
295 // Why at run time: the names are run-time strings here; `track` and `outbox!` reject a
296 // repeated name at compile time.
297 assert!(
298 !self.records.contains(name),
299 "`{name}` is registered with the outbox twice; register each name once"
300 );
301 Outbox {
302 store: self.store,
303 records: Registered {
304 name,
305 defaults: defaults::<DB, Record>(),
306 rest: self.records,
307 record: PhantomData,
308 },
309 }
310 }
311
312 /// Tracks the messages published under `Name::NAME` with the record type `Record`. A name
313 /// tracked twice does not compile. The record's default statements are built here, once.
314 ///
315 /// # Panics
316 ///
317 /// Panics when the name is registered already through [`register`](Self::register), whose
318 /// names the compiler does not see, and when `Record`'s table does not fit a dialect of `DB`
319 /// (an identifier longer than the database takes).
320 ///
321 /// # Examples
322 ///
323 /// ```
324 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
325 /// # mod demo {
326 /// # use ruststream::OutgoingMessage;
327 /// # use ruststream::memory::prelude::*;
328 /// # use ruststream_sqlx::{Outbox, outbox};
329 /// # use serde::{Deserialize, Serialize};
330 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
331 /// # #[derive(Outbox, sqlx::FromRow)]
332 /// # #[outbox(table = "outbox")]
333 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
334 /// # impl outbox::Publish<Postgres> for OrderOutbox {
335 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
336 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
337 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
338 /// # }
339 /// # }
340 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
341 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
342 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "refunds")] pub struct OrderCancelled { id: u64 }
343 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
344 /// # #[subscriber("cancel", reply)] async fn cancel(cmd: &PlaceOrder) -> OrderCancelled { OrderCancelled { id: cmd.id } }
345 /// use ruststream_sqlx::outbox::TrackedName;
346 ///
347 /// pub struct Orders;
348 ///
349 /// impl TrackedName for Orders {
350 /// const NAME: &'static str = "orders";
351 /// }
352 ///
353 /// pub struct Refunds;
354 ///
355 /// impl TrackedName for Refunds {
356 /// const NAME: &'static str = "refunds";
357 /// }
358 ///
359 /// /// One record type tracks both names; a third `track::<OrderOutbox, Orders>()` would not
360 /// /// compile.
361 /// pub fn app(pool: PgPool) -> impl App {
362 /// let tracking = Outbox::new(pool)
363 /// .track::<OrderOutbox, Orders>()
364 /// .track::<OrderOutbox, Refunds>();
365 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
366 /// .publish_layer(tracking.publish_layer())
367 /// .with_broker(MemoryBroker::new(), |b| {
368 /// b.include(place).out_reply(Publish);
369 /// b.include(cancel).out_reply(Publish);
370 /// b.after_startup(Publish, tracking.republish());
371 /// })
372 /// }
373 /// # }
374 /// # fn main() {}
375 /// ```
376 ///
377 /// A name tracked twice stops the build:
378 ///
379 /// ```compile_fail,E0080
380 /// # use ruststream::OutgoingMessage;
381 /// # use ruststream_sqlx::{Outbox, outbox};
382 /// # use sqlx::{Sqlite, SqliteConnection};
383 /// # #[derive(Outbox, sqlx::FromRow)]
384 /// # #[outbox(table = "outbox")]
385 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
386 /// # impl outbox::Publish<Sqlite> for OrderOutbox {
387 /// # async fn publish(conn: &mut SqliteConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
388 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
389 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
390 /// # }
391 /// # }
392 /// use ruststream_sqlx::outbox::TrackedName;
393 ///
394 /// pub struct Orders;
395 ///
396 /// impl TrackedName for Orders {
397 /// const NAME: &'static str = "orders";
398 /// }
399 ///
400 /// pub struct OrdersAgain;
401 ///
402 /// impl TrackedName for OrdersAgain {
403 /// const NAME: &'static str = "orders";
404 /// }
405 ///
406 /// fn main() {
407 /// let _ = Outbox::<Sqlite>::deferred()
408 /// .track::<OrderOutbox, Orders>()
409 /// .track::<OrderOutbox, OrdersAgain>();
410 /// }
411 /// ```
412 #[must_use]
413 #[track_caller]
414 pub fn track<Record: OutboxTable, Name: TrackedName>(
415 self,
416 ) -> Outbox<DB, Registered<Record, Checked<Name, Records>>>
417 where
418 DB: OutboxDatabase,
419 Records: Lacks<Name>,
420 Registered<Record, Checked<Name, Records>>: RecordList<DB>,
421 {
422 // The message cannot carry the name: a `const` panic formats only literals. The build
423 // names the call that repeats it, `track::<Record, Name>`.
424 const {
425 assert!(
426 <Records as Lacks<Name>>::LACKS,
427 "a name is tracked with the outbox twice: track each name once"
428 );
429 }
430 // Why at run time: a name registered through `register` is a run-time string.
431 assert!(
432 !self.records.contains(Name::NAME),
433 "`{}` is registered with the outbox twice; register each name once",
434 Name::NAME
435 );
436 Outbox {
437 store: self.store,
438 records: Registered {
439 name: Name::NAME,
440 defaults: defaults::<DB, Record>(),
441 rest: Checked(self.records, PhantomData),
442 record: PhantomData,
443 },
444 }
445 }
446
447 /// Gives the outbox its pool, when it was made with [`deferred`](Outbox::deferred).
448 ///
449 /// # Errors
450 ///
451 /// [`PoolAlreadySet`] when the outbox has a pool already; the one it has stays.
452 ///
453 /// # Examples
454 ///
455 /// ```
456 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
457 /// # mod demo {
458 /// # use ruststream::OutgoingMessage;
459 /// # use ruststream::memory::prelude::*;
460 /// # use ruststream_sqlx::{Outbox, outbox};
461 /// # use serde::{Deserialize, Serialize};
462 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
463 /// # #[derive(Outbox, sqlx::FromRow)]
464 /// # #[outbox(table = "outbox")]
465 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
466 /// # impl outbox::Publish<Postgres> for OrderOutbox {
467 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
468 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
469 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
470 /// # }
471 /// # }
472 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
473 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
474 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
475 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
476 /// use std::io;
477 ///
478 /// pub fn app() -> impl App {
479 /// let tracking = outbox! { "orders" => OrderOutbox };
480 /// let registry = tracking.clone();
481 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
482 /// .on_startup(async move |()| {
483 /// let pool = PgPool::connect("postgres://localhost/orders")
484 /// .await
485 /// .map_err(io::Error::other)?;
486 /// // Every handle the registry gave out reads this pool from now on.
487 /// registry.set_pool(pool.clone()).map_err(io::Error::other)?;
488 /// Ok::<_, io::Error>(pool)
489 /// })
490 /// .publish_layer(tracking.publish_layer())
491 /// .with_broker(MemoryBroker::new(), |b| {
492 /// b.include(place).out_reply(Publish);
493 /// })
494 /// }
495 /// # }
496 /// # fn main() {}
497 /// ```
498 pub fn set_pool(&self, pool: Pool<DB>) -> Result<(), PoolAlreadySet> {
499 // Why at run time: a pool is built asynchronously, often in `on_startup`, after the
500 // registry and its handles were made.
501 self.store.set(pool).map_err(|_| PoolAlreadySet)
502 }
503}
504
505impl<DB: Database, Records: RecordList<DB>> Outbox<DB, Records> {
506 /// The subscription middleware: a delivery under a registered name that carries an id takes
507 /// its record before the handler runs, and settles it by the handler's outcome.
508 ///
509 /// Mounted with `RustStream::layer`.
510 ///
511 /// # Examples
512 ///
513 /// ```
514 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
515 /// # mod demo {
516 /// # use ruststream::OutgoingMessage;
517 /// # use ruststream::memory::prelude::*;
518 /// # use ruststream_sqlx::{Outbox, outbox};
519 /// # use serde::{Deserialize, Serialize};
520 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
521 /// # #[derive(Outbox, sqlx::FromRow)]
522 /// # #[outbox(table = "outbox")]
523 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
524 /// # impl outbox::Publish<Postgres> for OrderOutbox {
525 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
526 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
527 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
528 /// # }
529 /// # }
530 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
531 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
532 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
533 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
534 /// /// The consumer side: `fulfil` runs once per record, and its acknowledgement marks the record.
535 /// pub fn app(pool: PgPool) -> impl App {
536 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
537 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
538 /// .layer(tracking.layer())
539 /// .with_broker(MemoryBroker::new(), |b| {
540 /// b.include(fulfil);
541 /// })
542 /// }
543 /// # }
544 /// # fn main() {}
545 /// ```
546 #[must_use]
547 pub fn layer(&self) -> TrackingLayer<DB, Records> {
548 TrackingLayer::new(Arc::clone(&self.store), self.records)
549 }
550
551 /// The publish middleware: a message a handler publishes under a registered name is recorded
552 /// before it is sent, and carries its record's id.
553 ///
554 /// Mounted with `RustStream::publish_layer`.
555 ///
556 /// # Examples
557 ///
558 /// ```
559 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
560 /// # mod demo {
561 /// # use ruststream::OutgoingMessage;
562 /// # use ruststream::memory::prelude::*;
563 /// # use ruststream_sqlx::{Outbox, outbox};
564 /// # use serde::{Deserialize, Serialize};
565 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
566 /// # #[derive(Outbox, sqlx::FromRow)]
567 /// # #[outbox(table = "outbox")]
568 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
569 /// # impl outbox::Publish<Postgres> for OrderOutbox {
570 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
571 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
572 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
573 /// # }
574 /// # }
575 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
576 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
577 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
578 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
579 /// /// The producer side: the reply of `place` is recorded, then sent with its record's id.
580 /// pub fn app(pool: PgPool) -> impl App {
581 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
582 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
583 /// .publish_layer(tracking.publish_layer())
584 /// .with_broker(MemoryBroker::new(), |b| {
585 /// b.include(place).out_reply(Publish);
586 /// })
587 /// }
588 /// # }
589 /// # fn main() {}
590 /// ```
591 #[must_use]
592 pub fn publish_layer(&self) -> TrackingPublishLayer<DB, Records> {
593 TrackingPublishLayer::new(Arc::clone(&self.store), self.records)
594 }
595
596 /// The startup republish of every registered name: an `after_startup` hook that publishes
597 /// each unprocessed record again through the scope's publisher, with its id.
598 ///
599 /// The hook allocates its body once, at startup.
600 ///
601 /// # Examples
602 ///
603 /// ```
604 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
605 /// # mod demo {
606 /// # use ruststream::OutgoingMessage;
607 /// # use ruststream::memory::prelude::*;
608 /// # use ruststream_sqlx::{Outbox, outbox};
609 /// # use serde::{Deserialize, Serialize};
610 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
611 /// # #[derive(Outbox, sqlx::FromRow)]
612 /// # #[outbox(table = "outbox")]
613 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
614 /// # impl outbox::Publish<Postgres> for OrderOutbox {
615 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
616 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
617 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
618 /// # }
619 /// # }
620 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
621 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
622 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
623 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
624 /// pub fn app(pool: PgPool) -> impl App {
625 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
626 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
627 /// .publish_layer(tracking.publish_layer())
628 /// .with_broker(MemoryBroker::new(), |b| {
629 /// b.include(place).out_reply(Publish);
630 /// // Runs once the subscriptions are open; a record it cannot send fails startup.
631 /// b.after_startup(Publish, tracking.republish());
632 /// })
633 /// }
634 /// # }
635 /// # fn main() {}
636 /// ```
637 pub fn republish<Live: Publisher + 'static>(
638 &self,
639 ) -> impl FnOnce(Live) -> Republishing + Send + 'static {
640 self.republishing(None)
641 }
642
643 /// The startup republish of the registered names `names` alone; see
644 /// [`republish`](Self::republish).
645 ///
646 /// # Panics
647 ///
648 /// Panics when a name of `names` is not registered.
649 ///
650 /// # Examples
651 ///
652 /// ```
653 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
654 /// # mod demo {
655 /// # use ruststream::OutgoingMessage;
656 /// # use ruststream::memory::prelude::*;
657 /// # use ruststream_sqlx::{Outbox, outbox};
658 /// # use serde::{Deserialize, Serialize};
659 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
660 /// # #[derive(Outbox, sqlx::FromRow)]
661 /// # #[outbox(table = "outbox")]
662 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
663 /// # impl outbox::Publish<Postgres> for OrderOutbox {
664 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
665 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
666 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
667 /// # }
668 /// # }
669 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
670 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
671 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
672 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
673 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "audits")] pub struct Audited { id: u64 }
674 /// # #[subscriber("checkout-audit", reply)] async fn audit(cmd: &PlaceOrder) -> Audited { Audited { id: cmd.id } }
675 /// /// Each broker republishes the names it carries.
676 /// pub fn app(pool: PgPool) -> impl App {
677 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox, "audits" => OrderOutbox };
678 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
679 /// .publish_layer(tracking.publish_layer())
680 /// .with_broker(MemoryBroker::new(), |b| {
681 /// b.include(place).out_reply(Publish);
682 /// b.after_startup(Publish, tracking.republish_names(["orders"]));
683 /// })
684 /// .with_broker(MemoryBroker::new(), |b| {
685 /// b.include(audit).out_reply(Publish);
686 /// b.after_startup(Publish, tracking.republish_names(["audits"]));
687 /// })
688 /// }
689 /// # }
690 /// # fn main() {}
691 /// ```
692 #[track_caller]
693 pub fn republish_names<Live: Publisher + 'static>(
694 &self,
695 names: impl IntoIterator<Item = &'static str>,
696 ) -> impl FnOnce(Live) -> Republishing + Send + 'static {
697 let names: Vec<&'static str> = names.into_iter().collect();
698 for name in &names {
699 assert!(
700 self.records.contains(name),
701 "`{name}` is not registered with the outbox, so it has nothing to republish"
702 );
703 }
704 self.republishing(Some(names))
705 }
706
707 fn republishing<Live: Publisher + 'static>(
708 &self,
709 only: Option<Vec<&'static str>>,
710 ) -> impl FnOnce(Live) -> Republishing + Send + 'static {
711 let pool = Arc::clone(&self.store);
712 let records = self.records;
713 move |publisher: Live| {
714 Republishing::new(
715 async move { records.republish(&pool, &publisher, only.as_deref()).await },
716 )
717 }
718 }
719
720 /// `publisher` with the publish middleware's tracking, for publishes outside the handlers
721 /// (an HTTP endpoint, an `after_startup` hook), which the publish pipeline does not reach.
722 ///
723 /// # Examples
724 ///
725 /// ```
726 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
727 /// # mod demo {
728 /// # use ruststream::OutgoingMessage;
729 /// # use ruststream::memory::prelude::*;
730 /// # use ruststream_sqlx::{Outbox, outbox};
731 /// # use serde::{Deserialize, Serialize};
732 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
733 /// # #[derive(Outbox, sqlx::FromRow)]
734 /// # #[outbox(table = "outbox")]
735 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
736 /// # impl outbox::Publish<Postgres> for OrderOutbox {
737 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
738 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
739 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
740 /// # }
741 /// # }
742 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
743 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
744 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
745 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
746 /// pub async fn serve(pool: PgPool) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
747 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
748 /// let broker = MemoryBroker::new().bindable();
749 /// let egress = broker.bind(Publish);
750 /// let running = RustStream::new(AppInfo::new("orders", "0.1.0"))
751 /// .with_broker(broker, |b| {
752 /// b.include(fulfil);
753 /// })
754 /// .start()
755 /// .await?;
756 ///
757 /// // What an HTTP endpoint publishes: recorded first, then sent with its record's id.
758 /// let publisher = tracking.wrap(running.publisher(egress).await?);
759 /// publisher.message(&OrderPlaced { id: 7 }).publish().await?;
760 ///
761 /// running.shutdown().await?;
762 /// Ok(())
763 /// }
764 /// # }
765 /// # fn main() {}
766 /// ```
767 #[must_use]
768 pub fn wrap<Live: Publisher>(&self, publisher: Live) -> TrackedPublisher<Live, DB, Records> {
769 TrackedPublisher::new(publisher, Arc::clone(&self.store), self.records)
770 }
771}