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`, for a name known
232 /// only at run time; [`track`](Self::track) refuses a repeated name while the service
233 /// compiles. The record's default statements are built here, once.
234 ///
235 /// # Panics
236 ///
237 /// Panics when `name` is registered already (each name has one record type), and when
238 /// `Record`'s table does not fit a dialect of `DB` (an identifier longer than the database
239 /// takes).
240 ///
241 /// # Examples
242 ///
243 /// ```
244 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
245 /// # mod demo {
246 /// # use ruststream::OutgoingMessage;
247 /// # use ruststream::memory::prelude::*;
248 /// # use ruststream_sqlx::{Outbox, outbox};
249 /// # use serde::{Deserialize, Serialize};
250 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
251 /// # #[derive(Outbox, sqlx::FromRow)]
252 /// # #[outbox(table = "outbox")]
253 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
254 /// # impl outbox::Publish<Postgres> for OrderOutbox {
255 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
256 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
257 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
258 /// # }
259 /// # }
260 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
261 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
262 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
263 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
264 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "refunds")] pub struct RefundRequested { id: u64 }
265 /// # #[subscriber("cancellations", reply)] async fn cancel(cmd: &PlaceOrder) -> RefundRequested { RefundRequested { id: cmd.id } }
266 /// /// One record type tracks both names; each name republishes its own records.
267 /// pub fn app(pool: PgPool) -> impl App {
268 /// let tracking = Outbox::new(pool)
269 /// .register::<OrderOutbox>("orders")
270 /// .register::<OrderOutbox>("refunds");
271 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
272 /// .publish_layer(tracking.publish_layer())
273 /// .with_broker(MemoryBroker::new(), |b| {
274 /// b.include(place).out_reply(Publish);
275 /// b.include(cancel).out_reply(Publish);
276 /// b.after_startup(Publish, tracking.republish());
277 /// })
278 /// }
279 /// # }
280 /// # fn main() {}
281 /// ```
282 #[must_use]
283 #[track_caller]
284 pub fn register<Record: OutboxTable>(
285 self,
286 name: &'static str,
287 ) -> Outbox<DB, Registered<Record, Records>>
288 where
289 DB: OutboxDatabase,
290 // On the node rather than on `Record`: rustc then reports the unmet bound underneath (a
291 // missing `Publish`, an event of the record's own without its impl) with its own message.
292 Registered<Record, Records>: RecordList<DB>,
293 {
294 // Why at run time: the names are run-time strings here; `track` and `outbox!` reject a
295 // repeated name at compile time.
296 assert!(
297 !self.records.contains(name),
298 "`{name}` is registered with the outbox twice; register each name once"
299 );
300 Outbox {
301 store: self.store,
302 records: Registered {
303 name,
304 defaults: defaults::<DB, Record>(),
305 rest: self.records,
306 record: PhantomData,
307 },
308 }
309 }
310
311 /// Tracks the messages published under `Name::NAME` with the record type `Record`. A name
312 /// tracked twice does not compile. The record's default statements are built here, once.
313 ///
314 /// # Panics
315 ///
316 /// Panics when the name is registered already through [`register`](Self::register), whose
317 /// names the compiler does not see, and when `Record`'s table does not fit a dialect of `DB`
318 /// (an identifier longer than the database takes).
319 ///
320 /// # Examples
321 ///
322 /// ```
323 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
324 /// # mod demo {
325 /// # use ruststream::OutgoingMessage;
326 /// # use ruststream::memory::prelude::*;
327 /// # use ruststream_sqlx::{Outbox, outbox};
328 /// # use serde::{Deserialize, Serialize};
329 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
330 /// # #[derive(Outbox, sqlx::FromRow)]
331 /// # #[outbox(table = "outbox")]
332 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
333 /// # impl outbox::Publish<Postgres> for OrderOutbox {
334 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
335 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
336 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
337 /// # }
338 /// # }
339 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
340 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
341 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "refunds")] pub struct OrderCancelled { id: u64 }
342 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
343 /// # #[subscriber("cancel", reply)] async fn cancel(cmd: &PlaceOrder) -> OrderCancelled { OrderCancelled { id: cmd.id } }
344 /// use ruststream_sqlx::outbox::TrackedName;
345 ///
346 /// pub struct Orders;
347 ///
348 /// impl TrackedName for Orders {
349 /// const NAME: &'static str = "orders";
350 /// }
351 ///
352 /// pub struct Refunds;
353 ///
354 /// impl TrackedName for Refunds {
355 /// const NAME: &'static str = "refunds";
356 /// }
357 ///
358 /// /// One record type tracks both names; a third `track::<OrderOutbox, Orders>()` would not
359 /// /// compile.
360 /// pub fn app(pool: PgPool) -> impl App {
361 /// let tracking = Outbox::new(pool)
362 /// .track::<OrderOutbox, Orders>()
363 /// .track::<OrderOutbox, Refunds>();
364 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
365 /// .publish_layer(tracking.publish_layer())
366 /// .with_broker(MemoryBroker::new(), |b| {
367 /// b.include(place).out_reply(Publish);
368 /// b.include(cancel).out_reply(Publish);
369 /// b.after_startup(Publish, tracking.republish());
370 /// })
371 /// }
372 /// # }
373 /// # fn main() {}
374 /// ```
375 ///
376 /// A name tracked twice stops the build:
377 ///
378 /// ```compile_fail,E0080
379 /// # use ruststream::OutgoingMessage;
380 /// # use ruststream_sqlx::{Outbox, outbox};
381 /// # use sqlx::{Sqlite, SqliteConnection};
382 /// # #[derive(Outbox, sqlx::FromRow)]
383 /// # #[outbox(table = "outbox")]
384 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
385 /// # impl outbox::Publish<Sqlite> for OrderOutbox {
386 /// # async fn publish(conn: &mut SqliteConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
387 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
388 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
389 /// # }
390 /// # }
391 /// use ruststream_sqlx::outbox::TrackedName;
392 ///
393 /// pub struct Orders;
394 ///
395 /// impl TrackedName for Orders {
396 /// const NAME: &'static str = "orders";
397 /// }
398 ///
399 /// pub struct OrdersAgain;
400 ///
401 /// impl TrackedName for OrdersAgain {
402 /// const NAME: &'static str = "orders";
403 /// }
404 ///
405 /// fn main() {
406 /// let _ = Outbox::<Sqlite>::deferred()
407 /// .track::<OrderOutbox, Orders>()
408 /// .track::<OrderOutbox, OrdersAgain>();
409 /// }
410 /// ```
411 #[must_use]
412 #[track_caller]
413 pub fn track<Record: OutboxTable, Name: TrackedName>(
414 self,
415 ) -> Outbox<DB, Registered<Record, Checked<Name, Records>>>
416 where
417 DB: OutboxDatabase,
418 Records: Lacks<Name>,
419 Registered<Record, Checked<Name, Records>>: RecordList<DB>,
420 {
421 // The message cannot carry the name: a `const` panic formats only literals. The build
422 // names the call that repeats it, `track::<Record, Name>`.
423 const {
424 assert!(
425 <Records as Lacks<Name>>::LACKS,
426 "a name is tracked with the outbox twice: track each name once"
427 );
428 }
429 // Why at run time: a name registered through `register` is a run-time string.
430 assert!(
431 !self.records.contains(Name::NAME),
432 "`{}` is registered with the outbox twice; register each name once",
433 Name::NAME
434 );
435 Outbox {
436 store: self.store,
437 records: Registered {
438 name: Name::NAME,
439 defaults: defaults::<DB, Record>(),
440 rest: Checked(self.records, PhantomData),
441 record: PhantomData,
442 },
443 }
444 }
445
446 /// Gives the outbox its pool, when it was made with [`deferred`](Outbox::deferred).
447 ///
448 /// # Errors
449 ///
450 /// [`PoolAlreadySet`] when the outbox has a pool already; the one it has stays.
451 ///
452 /// # Examples
453 ///
454 /// ```
455 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
456 /// # mod demo {
457 /// # use ruststream::OutgoingMessage;
458 /// # use ruststream::memory::prelude::*;
459 /// # use ruststream_sqlx::{Outbox, outbox};
460 /// # use serde::{Deserialize, Serialize};
461 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
462 /// # #[derive(Outbox, sqlx::FromRow)]
463 /// # #[outbox(table = "outbox")]
464 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
465 /// # impl outbox::Publish<Postgres> for OrderOutbox {
466 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
467 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
468 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
469 /// # }
470 /// # }
471 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
472 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
473 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
474 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
475 /// use std::io;
476 ///
477 /// pub fn app() -> impl App {
478 /// let tracking = outbox! { "orders" => OrderOutbox };
479 /// let registry = tracking.clone();
480 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
481 /// .on_startup(async move |()| {
482 /// let pool = PgPool::connect("postgres://localhost/orders")
483 /// .await
484 /// .map_err(io::Error::other)?;
485 /// // Every handle the registry gave out reads this pool from now on.
486 /// registry.set_pool(pool.clone()).map_err(io::Error::other)?;
487 /// Ok::<_, io::Error>(pool)
488 /// })
489 /// .publish_layer(tracking.publish_layer())
490 /// .with_broker(MemoryBroker::new(), |b| {
491 /// b.include(place).out_reply(Publish);
492 /// })
493 /// }
494 /// # }
495 /// # fn main() {}
496 /// ```
497 pub fn set_pool(&self, pool: Pool<DB>) -> Result<(), PoolAlreadySet> {
498 // Why at run time: a pool is built asynchronously, often in `on_startup`, after the
499 // registry and its handles were made.
500 self.store.set(pool).map_err(|_| PoolAlreadySet)
501 }
502}
503
504impl<DB: Database, Records: RecordList<DB>> Outbox<DB, Records> {
505 /// The subscription middleware: a delivery under a registered name that carries an id takes
506 /// its record before the handler runs, and settles it by the handler's outcome.
507 ///
508 /// Mounted with `RustStream::layer`.
509 ///
510 /// # Examples
511 ///
512 /// ```
513 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
514 /// # mod demo {
515 /// # use ruststream::OutgoingMessage;
516 /// # use ruststream::memory::prelude::*;
517 /// # use ruststream_sqlx::{Outbox, outbox};
518 /// # use serde::{Deserialize, Serialize};
519 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
520 /// # #[derive(Outbox, sqlx::FromRow)]
521 /// # #[outbox(table = "outbox")]
522 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
523 /// # impl outbox::Publish<Postgres> for OrderOutbox {
524 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
525 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
526 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
527 /// # }
528 /// # }
529 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
530 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
531 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
532 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
533 /// /// The consumer side: `fulfil` runs once per record, and its acknowledgement marks the record.
534 /// pub fn app(pool: PgPool) -> impl App {
535 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
536 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
537 /// .layer(tracking.layer())
538 /// .with_broker(MemoryBroker::new(), |b| {
539 /// b.include(fulfil);
540 /// })
541 /// }
542 /// # }
543 /// # fn main() {}
544 /// ```
545 #[must_use]
546 pub fn layer(&self) -> TrackingLayer<DB, Records> {
547 TrackingLayer::new(Arc::clone(&self.store), self.records)
548 }
549
550 /// The publish middleware: a message a handler publishes under a registered name is recorded
551 /// before it is sent, and carries its record's id.
552 ///
553 /// Mounted with `RustStream::publish_layer`.
554 ///
555 /// # Examples
556 ///
557 /// ```
558 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
559 /// # mod demo {
560 /// # use ruststream::OutgoingMessage;
561 /// # use ruststream::memory::prelude::*;
562 /// # use ruststream_sqlx::{Outbox, outbox};
563 /// # use serde::{Deserialize, Serialize};
564 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
565 /// # #[derive(Outbox, sqlx::FromRow)]
566 /// # #[outbox(table = "outbox")]
567 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
568 /// # impl outbox::Publish<Postgres> for OrderOutbox {
569 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
570 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
571 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
572 /// # }
573 /// # }
574 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
575 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
576 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
577 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
578 /// /// The producer side: the reply of `place` is recorded, then sent with its record's id.
579 /// pub fn app(pool: PgPool) -> impl App {
580 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
581 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
582 /// .publish_layer(tracking.publish_layer())
583 /// .with_broker(MemoryBroker::new(), |b| {
584 /// b.include(place).out_reply(Publish);
585 /// })
586 /// }
587 /// # }
588 /// # fn main() {}
589 /// ```
590 #[must_use]
591 pub fn publish_layer(&self) -> TrackingPublishLayer<DB, Records> {
592 TrackingPublishLayer::new(Arc::clone(&self.store), self.records)
593 }
594
595 /// The startup republish of every registered name: an `after_startup` hook that publishes
596 /// each unprocessed record again through the scope's publisher, with its id.
597 ///
598 /// The hook allocates its body once, at startup.
599 ///
600 /// # Examples
601 ///
602 /// ```
603 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
604 /// # mod demo {
605 /// # use ruststream::OutgoingMessage;
606 /// # use ruststream::memory::prelude::*;
607 /// # use ruststream_sqlx::{Outbox, outbox};
608 /// # use serde::{Deserialize, Serialize};
609 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
610 /// # #[derive(Outbox, sqlx::FromRow)]
611 /// # #[outbox(table = "outbox")]
612 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
613 /// # impl outbox::Publish<Postgres> for OrderOutbox {
614 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
615 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
616 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
617 /// # }
618 /// # }
619 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
620 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
621 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
622 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
623 /// pub fn app(pool: PgPool) -> impl App {
624 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
625 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
626 /// .publish_layer(tracking.publish_layer())
627 /// .with_broker(MemoryBroker::new(), |b| {
628 /// b.include(place).out_reply(Publish);
629 /// // Runs once the subscriptions are open; a record it cannot send fails startup.
630 /// b.after_startup(Publish, tracking.republish());
631 /// })
632 /// }
633 /// # }
634 /// # fn main() {}
635 /// ```
636 pub fn republish<Live: Publisher + 'static>(
637 &self,
638 ) -> impl FnOnce(Live) -> Republishing + Send + 'static {
639 self.republishing(None)
640 }
641
642 /// The startup republish of the registered names `names` alone; see
643 /// [`republish`](Self::republish).
644 ///
645 /// # Panics
646 ///
647 /// Panics when a name of `names` is not registered.
648 ///
649 /// # Examples
650 ///
651 /// ```
652 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
653 /// # mod demo {
654 /// # use ruststream::OutgoingMessage;
655 /// # use ruststream::memory::prelude::*;
656 /// # use ruststream_sqlx::{Outbox, outbox};
657 /// # use serde::{Deserialize, Serialize};
658 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
659 /// # #[derive(Outbox, sqlx::FromRow)]
660 /// # #[outbox(table = "outbox")]
661 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
662 /// # impl outbox::Publish<Postgres> for OrderOutbox {
663 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
664 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
665 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
666 /// # }
667 /// # }
668 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
669 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
670 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
671 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
672 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "audits")] pub struct Audited { id: u64 }
673 /// # #[subscriber("checkout-audit", reply)] async fn audit(cmd: &PlaceOrder) -> Audited { Audited { id: cmd.id } }
674 /// /// Each broker republishes the names it carries.
675 /// pub fn app(pool: PgPool) -> impl App {
676 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox, "audits" => OrderOutbox };
677 /// RustStream::new(AppInfo::new("orders", "0.1.0"))
678 /// .publish_layer(tracking.publish_layer())
679 /// .with_broker(MemoryBroker::new(), |b| {
680 /// b.include(place).out_reply(Publish);
681 /// b.after_startup(Publish, tracking.republish_names(["orders"]));
682 /// })
683 /// .with_broker(MemoryBroker::new(), |b| {
684 /// b.include(audit).out_reply(Publish);
685 /// b.after_startup(Publish, tracking.republish_names(["audits"]));
686 /// })
687 /// }
688 /// # }
689 /// # fn main() {}
690 /// ```
691 #[track_caller]
692 pub fn republish_names<Live: Publisher + 'static>(
693 &self,
694 names: impl IntoIterator<Item = &'static str>,
695 ) -> impl FnOnce(Live) -> Republishing + Send + 'static {
696 let names: Vec<&'static str> = names.into_iter().collect();
697 for name in &names {
698 assert!(
699 self.records.contains(name),
700 "`{name}` is not registered with the outbox, so it has nothing to republish"
701 );
702 }
703 self.republishing(Some(names))
704 }
705
706 fn republishing<Live: Publisher + 'static>(
707 &self,
708 only: Option<Vec<&'static str>>,
709 ) -> impl FnOnce(Live) -> Republishing + Send + 'static {
710 let pool = Arc::clone(&self.store);
711 let records = self.records;
712 move |publisher: Live| {
713 Republishing::new(
714 async move { records.republish(&pool, &publisher, only.as_deref()).await },
715 )
716 }
717 }
718
719 /// `publisher` with the publish middleware's tracking, for publishes outside the handlers
720 /// (an HTTP endpoint, an `after_startup` hook), which the publish pipeline does not reach.
721 ///
722 /// # Examples
723 ///
724 /// ```
725 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
726 /// # mod demo {
727 /// # use ruststream::OutgoingMessage;
728 /// # use ruststream::memory::prelude::*;
729 /// # use ruststream_sqlx::{Outbox, outbox};
730 /// # use serde::{Deserialize, Serialize};
731 /// # use sqlx::postgres::{PgConnection, PgPool, Postgres};
732 /// # #[derive(Outbox, sqlx::FromRow)]
733 /// # #[outbox(table = "outbox")]
734 /// # pub struct OrderOutbox { #[field(id)] id: i64, #[field(name)] name: String, #[field(payload)] payload: Vec<u8> }
735 /// # impl outbox::Publish<Postgres> for OrderOutbox {
736 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
737 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
738 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
739 /// # }
740 /// # }
741 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
742 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
743 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
744 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
745 /// pub async fn serve(pool: PgPool) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
746 /// let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
747 /// let broker = MemoryBroker::new().bindable();
748 /// let egress = broker.bind(Publish);
749 /// let running = RustStream::new(AppInfo::new("orders", "0.1.0"))
750 /// .with_broker(broker, |b| {
751 /// b.include(fulfil);
752 /// })
753 /// .start()
754 /// .await?;
755 ///
756 /// // What an HTTP endpoint publishes: recorded first, then sent with its record's id.
757 /// let publisher = tracking.wrap(running.publisher(egress).await?);
758 /// publisher.message(&OrderPlaced { id: 7 }).publish().await?;
759 ///
760 /// running.shutdown().await?;
761 /// Ok(())
762 /// }
763 /// # }
764 /// # fn main() {}
765 /// ```
766 #[must_use]
767 pub fn wrap<Live: Publisher>(&self, publisher: Live) -> TrackedPublisher<Live, DB, Records> {
768 TrackedPublisher::new(publisher, Arc::clone(&self.store), self.records)
769 }
770}