ruststream_sqlx/outbox/spec.rs
1//! The settings of an outbox table described by hand, as types.
2//!
3//! Each typed setter of [`OutboxSpec`] adds one marker of this module to the
4//! table's settings, and the table's `type Table` lists them in the order the chain sets them. A
5//! setting the chain leaves out keeps its default: no headers column, a processed record deleted,
6//! and the crate's events. A setting set twice does not compile ([`Merge`]).
7
8mod builder;
9mod declaration;
10
11pub use builder::{Described, OutboxSpec, OutboxTable};
12pub use declaration::Declaration;
13use declaration::setting;
14
15pub use crate::settings::{Merge, Push, Set, Unset};
16
17setting!(
18 /// The published headers, kept in one column: set by
19 /// [`OutboxSpec::headers`](super::OutboxSpec::headers). The record implements
20 /// [`HeaderRow`](crate::HeaderRow).
21 ///
22 /// # Examples
23 ///
24 /// ```
25 /// # #[cfg(all(feature = "outbox", feature = "json", feature = "postgres"))]
26 /// # mod demo {
27 /// # use ruststream::OutgoingMessage;
28 /// # use ruststream::memory::prelude::*;
29 /// # use serde::{Deserialize, Serialize};
30 /// # use ruststream_sqlx::outbox::{Outbox, TrackedName};
31 /// use std::collections::BTreeMap;
32 ///
33 /// use ruststream_sqlx::HeaderRow;
34 /// use ruststream_sqlx::dialect::Column;
35 /// use ruststream_sqlx::outbox::spec::Headers;
36 /// use ruststream_sqlx::outbox::{self, OutboxSpec, OutboxTable};
37 /// use sqlx::types::Json;
38 /// use sqlx::{PgConnection, PgPool, Postgres};
39 ///
40 /// #[derive(sqlx::FromRow)]
41 /// pub struct OrderEvent {
42 /// id: i64,
43 /// name: String,
44 /// payload: Vec<u8>,
45 /// headers: Option<Json<BTreeMap<String, String>>>,
46 /// }
47 ///
48 /// impl OutboxTable for OrderEvent {
49 /// type Id = i64;
50 /// type Table = OutboxSpec<(Headers,)>;
51 /// const TABLE: Self::Table = OutboxSpec::new(
52 /// "outbox",
53 /// Column::new("id"),
54 /// Column::new("name"),
55 /// Column::new("payload"),
56 /// )
57 /// .headers(Column::new("headers"));
58 /// # fn id(&self) -> &i64 { &self.id }
59 /// # fn name(&self) -> &str { &self.name }
60 /// # fn payload(&self) -> &[u8] { &self.payload }
61 /// }
62 ///
63 /// // A republished record carries the headers it was published with.
64 /// impl HeaderRow for OrderEvent {
65 /// type Column = Option<Json<BTreeMap<String, String>>>;
66 ///
67 /// fn headers_mut(&mut self) -> &mut Self::Column {
68 /// &mut self.headers
69 /// }
70 /// }
71 /// # impl outbox::Publish<Postgres> for OrderEvent {
72 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
73 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
74 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
75 /// # }
76 /// # }
77 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
78 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
79 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
80 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
81 /// # pub struct Orders;
82 /// # impl TrackedName for Orders { const NAME: &'static str = "orders"; }
83 /// # pub fn app(pool: PgPool) -> impl App {
84 /// # let tracking = Outbox::new(pool).track::<OrderEvent, Orders>();
85 /// # RustStream::new(AppInfo::new("orders", "0.1.0"))
86 /// # .layer(tracking.layer())
87 /// # .publish_layer(tracking.publish_layer())
88 /// # .with_broker(MemoryBroker::new(), |b| {
89 /// # b.include(place).out_reply(Publish);
90 /// # b.include(fulfil);
91 /// # b.after_startup(Publish, tracking.republish());
92 /// # })
93 /// # }
94 /// # }
95 /// # fn main() {}
96 /// ```
97 Headers, Headers
98);
99setting!(
100 /// The time a record was processed, which the database's clock writes: set by
101 /// [`OutboxSpec::processed_at`](super::OutboxSpec::processed_at). Without it a processed record
102 /// is deleted.
103 ///
104 /// # Examples
105 ///
106 /// ```
107 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
108 /// # mod demo {
109 /// # use ruststream::OutgoingMessage;
110 /// # use ruststream::memory::prelude::*;
111 /// # use serde::{Deserialize, Serialize};
112 /// # use ruststream_sqlx::outbox::{Outbox, TrackedName};
113 /// use ruststream_sqlx::dialect::Column;
114 /// use ruststream_sqlx::outbox::spec::ProcessedAt;
115 /// use ruststream_sqlx::outbox::{self, OutboxSpec, OutboxTable};
116 /// use sqlx::{PgConnection, PgPool, Postgres};
117 ///
118 /// #[derive(sqlx::FromRow)]
119 /// pub struct OrderEvent {
120 /// id: i64,
121 /// name: String,
122 /// payload: Vec<u8>,
123 /// }
124 ///
125 /// // A processed record stays in the table, stamped with the database's time.
126 /// impl OutboxTable for OrderEvent {
127 /// type Id = i64;
128 /// type Table = OutboxSpec<(ProcessedAt,)>;
129 /// const TABLE: Self::Table = OutboxSpec::new(
130 /// "outbox",
131 /// Column::new("id"),
132 /// Column::new("name"),
133 /// Column::new("payload"),
134 /// )
135 /// .processed_at(Column::new("processed_at"));
136 /// # fn id(&self) -> &i64 { &self.id }
137 /// # fn name(&self) -> &str { &self.name }
138 /// # fn payload(&self) -> &[u8] { &self.payload }
139 /// }
140 /// # impl outbox::Publish<Postgres> for OrderEvent {
141 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
142 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
143 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
144 /// # }
145 /// # }
146 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
147 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
148 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
149 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
150 /// # pub struct Orders;
151 /// # impl TrackedName for Orders { const NAME: &'static str = "orders"; }
152 /// # pub fn app(pool: PgPool) -> impl App {
153 /// # let tracking = Outbox::new(pool).track::<OrderEvent, Orders>();
154 /// # RustStream::new(AppInfo::new("orders", "0.1.0"))
155 /// # .layer(tracking.layer())
156 /// # .publish_layer(tracking.publish_layer())
157 /// # .with_broker(MemoryBroker::new(), |b| {
158 /// # b.include(place).out_reply(Publish);
159 /// # b.include(fulfil);
160 /// # b.after_startup(Publish, tracking.republish());
161 /// # })
162 /// # }
163 /// # }
164 /// # fn main() {}
165 /// ```
166 ProcessedAt, ProcessedAt
167);
168
169/// The events a record runs itself instead of the crate's default, each set by
170/// [`OutboxSpec::own`](super::OutboxSpec::own); the record implements the event's trait of
171/// [`outbox`](super).
172pub mod own {
173 use super::declaration::{Declaration, setting};
174 use super::{Set, Unset};
175
176 setting!(
177 /// The take of a record when its message reaches a subscription:
178 /// [`outbox::Fetch`](crate::outbox::Fetch).
179 ///
180 /// # Examples
181 ///
182 /// ```
183 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
184 /// # mod demo {
185 /// # use ruststream::OutgoingMessage;
186 /// # use ruststream::memory::prelude::*;
187 /// # use serde::{Deserialize, Serialize};
188 /// # use ruststream_sqlx::outbox::{Outbox, TrackedName};
189 /// use ruststream_sqlx::dialect::Column;
190 /// use ruststream_sqlx::outbox::spec::own;
191 /// use ruststream_sqlx::outbox::{self, OutboxSpec, OutboxTable};
192 /// use sqlx::{PgConnection, PgPool, Postgres};
193 ///
194 /// #[derive(sqlx::FromRow)]
195 /// pub struct OrderEvent {
196 /// id: i64,
197 /// name: String,
198 /// payload: Vec<u8>,
199 /// }
200 ///
201 /// impl OutboxTable for OrderEvent {
202 /// type Id = i64;
203 /// type Table = OutboxSpec<(own::Fetch,)>;
204 /// const TABLE: Self::Table = OutboxSpec::new(
205 /// "outbox",
206 /// Column::new("id"),
207 /// Column::new("name"),
208 /// Column::new("payload"),
209 /// )
210 /// .own::<own::Fetch>();
211 /// # fn id(&self) -> &i64 { &self.id }
212 /// # fn name(&self) -> &str { &self.name }
213 /// # fn payload(&self) -> &[u8] { &self.payload }
214 /// }
215 /// # impl outbox::Publish<Postgres> for OrderEvent {
216 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
217 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
218 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
219 /// # }
220 /// # }
221 ///
222 /// // A record the service cancelled is not delivered: its delivery is acknowledged without the handler.
223 /// impl outbox::Fetch<Postgres> for OrderEvent {
224 /// async fn fetch(conn: &mut PgConnection, id: &i64) -> sqlx::Result<Option<Self>> {
225 /// sqlx::query_as("SELECT id, name, payload FROM outbox WHERE id = $1 AND NOT cancelled")
226 /// .bind(id)
227 /// .fetch_optional(conn)
228 /// .await
229 /// }
230 /// }
231 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
232 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
233 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
234 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
235 /// # pub struct Orders;
236 /// # impl TrackedName for Orders { const NAME: &'static str = "orders"; }
237 /// # pub fn app(pool: PgPool) -> impl App {
238 /// # let tracking = Outbox::new(pool).track::<OrderEvent, Orders>();
239 /// # RustStream::new(AppInfo::new("orders", "0.1.0"))
240 /// # .layer(tracking.layer())
241 /// # .publish_layer(tracking.publish_layer())
242 /// # .with_broker(MemoryBroker::new(), |b| {
243 /// # b.include(place).out_reply(Publish);
244 /// # b.include(fulfil);
245 /// # b.after_startup(Publish, tracking.republish());
246 /// # })
247 /// # }
248 /// # }
249 /// # fn main() {}
250 /// ```
251 Fetch, OwnFetch
252 );
253 setting!(
254 /// The mark of an acknowledged record: [`outbox::Ack`](crate::outbox::Ack).
255 ///
256 /// # Examples
257 ///
258 /// ```
259 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
260 /// # mod demo {
261 /// # use ruststream::OutgoingMessage;
262 /// # use ruststream::memory::prelude::*;
263 /// # use serde::{Deserialize, Serialize};
264 /// # use ruststream_sqlx::outbox::{Outbox, TrackedName};
265 /// use ruststream_sqlx::dialect::Column;
266 /// use ruststream_sqlx::outbox::spec::own;
267 /// use ruststream_sqlx::outbox::{self, OutboxSpec, OutboxTable};
268 /// use sqlx::{PgConnection, PgPool, Postgres};
269 ///
270 /// #[derive(sqlx::FromRow)]
271 /// pub struct OrderEvent {
272 /// id: i64,
273 /// name: String,
274 /// payload: Vec<u8>,
275 /// }
276 ///
277 /// impl OutboxTable for OrderEvent {
278 /// type Id = i64;
279 /// type Table = OutboxSpec<(own::Ack,)>;
280 /// const TABLE: Self::Table = OutboxSpec::new(
281 /// "outbox",
282 /// Column::new("id"),
283 /// Column::new("name"),
284 /// Column::new("payload"),
285 /// )
286 /// .own::<own::Ack>();
287 /// # fn id(&self) -> &i64 { &self.id }
288 /// # fn name(&self) -> &str { &self.name }
289 /// # fn payload(&self) -> &[u8] { &self.payload }
290 /// }
291 /// # impl outbox::Publish<Postgres> for OrderEvent {
292 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
293 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
294 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
295 /// # }
296 /// # }
297 ///
298 /// // An acknowledged record moves to the archive.
299 /// impl outbox::Ack<Postgres> for OrderEvent {
300 /// async fn ack(conn: &mut PgConnection, id: &i64) -> sqlx::Result<()> {
301 /// sqlx::query(
302 /// "WITH done AS (DELETE FROM outbox WHERE id = $1 RETURNING *) \
303 /// INSERT INTO outbox_archive SELECT * FROM done",
304 /// )
305 /// .bind(id)
306 /// .execute(conn)
307 /// .await?;
308 /// Ok(())
309 /// }
310 /// }
311 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
312 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
313 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
314 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
315 /// # pub struct Orders;
316 /// # impl TrackedName for Orders { const NAME: &'static str = "orders"; }
317 /// # pub fn app(pool: PgPool) -> impl App {
318 /// # let tracking = Outbox::new(pool).track::<OrderEvent, Orders>();
319 /// # RustStream::new(AppInfo::new("orders", "0.1.0"))
320 /// # .layer(tracking.layer())
321 /// # .publish_layer(tracking.publish_layer())
322 /// # .with_broker(MemoryBroker::new(), |b| {
323 /// # b.include(place).out_reply(Publish);
324 /// # b.include(fulfil);
325 /// # b.after_startup(Publish, tracking.republish());
326 /// # })
327 /// # }
328 /// # }
329 /// # fn main() {}
330 /// ```
331 Ack, OwnAck
332 );
333 setting!(
334 /// What a retried record runs: [`outbox::Retry`](crate::outbox::Retry).
335 ///
336 /// # Examples
337 ///
338 /// ```
339 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
340 /// # mod demo {
341 /// # use ruststream::OutgoingMessage;
342 /// # use ruststream::memory::prelude::*;
343 /// # use serde::{Deserialize, Serialize};
344 /// # use ruststream_sqlx::outbox::{Outbox, TrackedName};
345 /// use ruststream_sqlx::dialect::Column;
346 /// use ruststream_sqlx::outbox::spec::own;
347 /// use ruststream_sqlx::outbox::{self, OutboxSpec, OutboxTable};
348 /// use sqlx::{PgConnection, PgPool, Postgres};
349 ///
350 /// #[derive(sqlx::FromRow)]
351 /// pub struct OrderEvent {
352 /// id: i64,
353 /// name: String,
354 /// payload: Vec<u8>,
355 /// }
356 ///
357 /// impl OutboxTable for OrderEvent {
358 /// type Id = i64;
359 /// type Table = OutboxSpec<(own::Retry,)>;
360 /// const TABLE: Self::Table = OutboxSpec::new(
361 /// "outbox",
362 /// Column::new("id"),
363 /// Column::new("name"),
364 /// Column::new("payload"),
365 /// )
366 /// .own::<own::Retry>();
367 /// # fn id(&self) -> &i64 { &self.id }
368 /// # fn name(&self) -> &str { &self.name }
369 /// # fn payload(&self) -> &[u8] { &self.payload }
370 /// }
371 /// # impl outbox::Publish<Postgres> for OrderEvent {
372 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
373 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
374 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
375 /// # }
376 /// # }
377 ///
378 /// // The service counts how often a consumer asked for a record again.
379 /// impl outbox::Retry<Postgres> for OrderEvent {
380 /// async fn retry(conn: &mut PgConnection, id: &i64) -> sqlx::Result<()> {
381 /// sqlx::query("UPDATE outbox SET retries = retries + 1 WHERE id = $1")
382 /// .bind(id)
383 /// .execute(conn)
384 /// .await?;
385 /// Ok(())
386 /// }
387 /// }
388 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
389 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
390 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
391 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
392 /// # pub struct Orders;
393 /// # impl TrackedName for Orders { const NAME: &'static str = "orders"; }
394 /// # pub fn app(pool: PgPool) -> impl App {
395 /// # let tracking = Outbox::new(pool).track::<OrderEvent, Orders>();
396 /// # RustStream::new(AppInfo::new("orders", "0.1.0"))
397 /// # .layer(tracking.layer())
398 /// # .publish_layer(tracking.publish_layer())
399 /// # .with_broker(MemoryBroker::new(), |b| {
400 /// # b.include(place).out_reply(Publish);
401 /// # b.include(fulfil);
402 /// # b.after_startup(Publish, tracking.republish());
403 /// # })
404 /// # }
405 /// # }
406 /// # fn main() {}
407 /// ```
408 Retry, OwnRetry
409 );
410 setting!(
411 /// The mark of a dropped record: [`outbox::Discard`](crate::outbox::Discard).
412 ///
413 /// # Examples
414 ///
415 /// ```
416 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
417 /// # mod demo {
418 /// # use ruststream::OutgoingMessage;
419 /// # use ruststream::memory::prelude::*;
420 /// # use serde::{Deserialize, Serialize};
421 /// # use ruststream_sqlx::outbox::{Outbox, TrackedName};
422 /// use ruststream_sqlx::dialect::Column;
423 /// use ruststream_sqlx::outbox::spec::own;
424 /// use ruststream_sqlx::outbox::{self, OutboxSpec, OutboxTable};
425 /// use sqlx::{PgConnection, PgPool, Postgres};
426 ///
427 /// #[derive(sqlx::FromRow)]
428 /// pub struct OrderEvent {
429 /// id: i64,
430 /// name: String,
431 /// payload: Vec<u8>,
432 /// }
433 ///
434 /// impl OutboxTable for OrderEvent {
435 /// type Id = i64;
436 /// type Table = OutboxSpec<(own::Discard,)>;
437 /// const TABLE: Self::Table = OutboxSpec::new(
438 /// "outbox",
439 /// Column::new("id"),
440 /// Column::new("name"),
441 /// Column::new("payload"),
442 /// )
443 /// .own::<own::Discard>();
444 /// # fn id(&self) -> &i64 { &self.id }
445 /// # fn name(&self) -> &str { &self.name }
446 /// # fn payload(&self) -> &[u8] { &self.payload }
447 /// }
448 /// # impl outbox::Publish<Postgres> for OrderEvent {
449 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
450 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
451 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
452 /// # }
453 /// # }
454 ///
455 /// // A dropped record moves to a table the operators review.
456 /// impl outbox::Discard<Postgres> for OrderEvent {
457 /// async fn discard(conn: &mut PgConnection, id: &i64) -> sqlx::Result<()> {
458 /// sqlx::query(
459 /// "WITH dropped AS (DELETE FROM outbox WHERE id = $1 RETURNING *) \
460 /// INSERT INTO outbox_dropped SELECT * FROM dropped",
461 /// )
462 /// .bind(id)
463 /// .execute(conn)
464 /// .await?;
465 /// Ok(())
466 /// }
467 /// }
468 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
469 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
470 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
471 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
472 /// # pub struct Orders;
473 /// # impl TrackedName for Orders { const NAME: &'static str = "orders"; }
474 /// # pub fn app(pool: PgPool) -> impl App {
475 /// # let tracking = Outbox::new(pool).track::<OrderEvent, Orders>();
476 /// # RustStream::new(AppInfo::new("orders", "0.1.0"))
477 /// # .layer(tracking.layer())
478 /// # .publish_layer(tracking.publish_layer())
479 /// # .with_broker(MemoryBroker::new(), |b| {
480 /// # b.include(place).out_reply(Publish);
481 /// # b.include(fulfil);
482 /// # b.after_startup(Publish, tracking.republish());
483 /// # })
484 /// # }
485 /// # }
486 /// # fn main() {}
487 /// ```
488 Discard, OwnDiscard
489 );
490 setting!(
491 /// The selection of the unprocessed records at startup:
492 /// [`outbox::Recover`](crate::outbox::Recover).
493 ///
494 /// # Examples
495 ///
496 /// ```
497 /// # #[cfg(all(feature = "outbox", feature = "postgres"))]
498 /// # mod demo {
499 /// # use ruststream::OutgoingMessage;
500 /// # use ruststream::memory::prelude::*;
501 /// # use serde::{Deserialize, Serialize};
502 /// # use ruststream_sqlx::outbox::{Outbox, TrackedName};
503 /// use ruststream_sqlx::dialect::Column;
504 /// use ruststream_sqlx::outbox::spec::own;
505 /// use ruststream_sqlx::outbox::{self, OutboxSpec, OutboxTable};
506 /// use sqlx::{PgConnection, PgPool, Postgres};
507 ///
508 /// #[derive(sqlx::FromRow)]
509 /// pub struct OrderEvent {
510 /// id: i64,
511 /// name: String,
512 /// payload: Vec<u8>,
513 /// }
514 ///
515 /// impl OutboxTable for OrderEvent {
516 /// type Id = i64;
517 /// type Table = OutboxSpec<(own::Recover,)>;
518 /// const TABLE: Self::Table = OutboxSpec::new(
519 /// "outbox",
520 /// Column::new("id"),
521 /// Column::new("name"),
522 /// Column::new("payload"),
523 /// )
524 /// .own::<own::Recover>();
525 /// # fn id(&self) -> &i64 { &self.id }
526 /// # fn name(&self) -> &str { &self.name }
527 /// # fn payload(&self) -> &[u8] { &self.payload }
528 /// }
529 /// # impl outbox::Publish<Postgres> for OrderEvent {
530 /// # async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
531 /// # sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
532 /// # .bind(msg.name()).bind(msg.payload()).fetch_one(conn).await
533 /// # }
534 /// # }
535 ///
536 /// // The startup republish sends the records oldest first.
537 /// impl outbox::Recover<Postgres> for OrderEvent {
538 /// async fn recover(conn: &mut PgConnection, name: &str) -> sqlx::Result<Vec<Self>> {
539 /// sqlx::query_as("SELECT id, name, payload FROM outbox WHERE name = $1 ORDER BY id")
540 /// .bind(name)
541 /// .fetch_all(conn)
542 /// .await
543 /// }
544 /// }
545 /// # #[derive(Deserialize)] pub struct PlaceOrder { id: u64 }
546 /// # #[derive(Serialize, Deserialize, Outgoing)] #[outgoing(name = "orders")] pub struct OrderPlaced { id: u64 }
547 /// # #[subscriber("checkout", reply)] async fn place(cmd: &PlaceOrder) -> OrderPlaced { OrderPlaced { id: cmd.id } }
548 /// # #[subscriber("orders")] async fn fulfil(_: &OrderPlaced) -> HandlerOutcome { HandlerOutcome::ack() }
549 /// # pub struct Orders;
550 /// # impl TrackedName for Orders { const NAME: &'static str = "orders"; }
551 /// # pub fn app(pool: PgPool) -> impl App {
552 /// # let tracking = Outbox::new(pool).track::<OrderEvent, Orders>();
553 /// # RustStream::new(AppInfo::new("orders", "0.1.0"))
554 /// # .layer(tracking.layer())
555 /// # .publish_layer(tracking.publish_layer())
556 /// # .with_broker(MemoryBroker::new(), |b| {
557 /// # b.include(place).out_reply(Publish);
558 /// # b.include(fulfil);
559 /// # b.after_startup(Publish, tracking.republish());
560 /// # })
561 /// # }
562 /// # }
563 /// # fn main() {}
564 /// ```
565 Recover, OwnRecover
566 );
567}
568
569/// An event a record runs itself: a marker of [`own`], which
570/// [`OutboxSpec::own`](super::OutboxSpec::own) takes.
571#[diagnostic::on_unimplemented(
572 message = "`{Self}` is not an event of an outbox record",
573 label = "not an outbox event",
574 note = "name one of `outbox::spec::own::{{Fetch, Ack, Retry, Discard, Recover}}`"
575)]
576pub trait OwnEvent: Declaration {}
577
578impl OwnEvent for own::Fetch {}
579impl OwnEvent for own::Ack {}
580impl OwnEvent for own::Retry {}
581impl OwnEvent for own::Discard {}
582impl OwnEvent for own::Recover {}