ruststream_sqlx/inbox/spec.rs
1//! The settings of a queue table described by hand, as types.
2//!
3//! Each typed setter of [`InboxSpec`](crate::InboxSpec) 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: the row lock form, row mode, the
6//! [`SystemClock`](crate::SystemClock), the database's default opening and the crate's events. A
7//! setting set twice does not compile ([`Merge`]), and neither does a combination of settings no
8//! table can hold.
9
10pub(super) mod builder;
11mod declaration;
12mod rules;
13
14use std::marker::PhantomData;
15
16use ruststream_sqlx_dialect::{Isolation, Mode, Opening, level};
17
18use declaration::setting;
19pub use declaration::{Declaration, Merge, Push, Set, Unset};
20#[doc(hidden)]
21pub use rules::{
22 ClaimOutsideAdvisory, ClockSlot, ExtendInLease, FifoOutsideAdvisory, LeaseOnServiceClock,
23 LockInAdvisory, LockWithUnlock, PayloadOutsideHeaderFields, Valid,
24};
25
26setting!(
27 /// The lease form, its expiry in `Time`: set by [`InboxSpec::lease`](crate::InboxSpec::lease).
28 ///
29 /// # Examples
30 ///
31 /// ```
32 /// # #[cfg(all(feature = "sqlite", feature = "chrono"))]
33 /// # mod demo {
34 /// use chrono::{DateTime, Utc};
35 /// use ruststream_sqlx::dialect::Column;
36 /// use ruststream_sqlx::spec::{Lease, Payload};
37 /// use ruststream_sqlx::{InboxSpec, InboxTable};
38 /// # use ruststream_sqlx::PayloadRow;
39 /// # use ruststream_sqlx::prelude::*;
40 /// # use serde::Deserialize;
41 /// # use sqlx::{Sqlite, SqlitePool};
42 ///
43 /// #[derive(sqlx::FromRow)]
44 /// pub struct ScanJob {
45 /// id: i64,
46 /// payload: Vec<u8>,
47 /// }
48 ///
49 /// impl InboxTable for ScanJob {
50 /// type Id = i64;
51 /// // A virus scan holds its row by a lease, kept in `locked_until` as a UTC timestamp.
52 /// type Table = InboxSpec<(Lease<DateTime<Utc>>, Payload)>;
53 /// const TABLE: Self::Table = InboxSpec::new("scan_jobs", Column::new("id").generated())
54 /// .lease(Column::new("locked_until"))
55 /// .payload(Column::new("payload"));
56 ///
57 /// fn id(&self) -> &i64 {
58 /// &self.id
59 /// }
60 /// }
61 /// # impl PayloadRow for ScanJob {
62 /// # type Column = Vec<u8>;
63 /// # fn payload(&self) -> &[u8] { &self.payload }
64 /// # }
65 /// # #[derive(Deserialize)]
66 /// # struct Upload { path: String }
67 /// # #[subscriber(InboxQueue::<ScanJob>::new("scans"))]
68 /// # async fn scan(upload: &Upload) -> HandlerOutcome {
69 /// # tracing::info!(path = %upload.path, "scanning");
70 /// # HandlerOutcome::ack()
71 /// # }
72 /// # pub fn app(pool: SqlitePool) -> RustStream {
73 /// # RustStream::new(AppInfo::new("scanner", "1.0.0"))
74 /// # .with_broker(SqlxBroker::<Sqlite>::new(pool), |b| {
75 /// # b.include(scan);
76 /// # })
77 /// # }
78 /// # }
79 /// # fn main() {}
80 /// ```
81 Lease<Time>, Form
82);
83setting!(
84 /// The advisory lock form: set by [`InboxSpec::advisory`](crate::InboxSpec::advisory).
85 ///
86 /// # Examples
87 ///
88 /// ```
89 /// # #[cfg(feature = "mysql")]
90 /// # mod demo {
91 /// use ruststream_sqlx::dialect::{Column, KeyPart};
92 /// use ruststream_sqlx::spec::{Advisory, Payload};
93 /// use ruststream_sqlx::{InboxSpec, InboxTable};
94 /// # use ruststream_sqlx::PayloadRow;
95 /// # use ruststream_sqlx::prelude::*;
96 /// # use serde::Deserialize;
97 /// # use sqlx::{MySql, MySqlPool};
98 ///
99 /// #[derive(sqlx::FromRow)]
100 /// pub struct FirmwareUpdate {
101 /// id: i64,
102 /// payload: Vec<u8>,
103 /// }
104 ///
105 /// impl InboxTable for FirmwareUpdate {
106 /// type Id = i64;
107 /// // A device takes one update at a time: the lock key is `firmware-<device>`.
108 /// type Table = InboxSpec<(Advisory, Payload)>;
109 /// const TABLE: Self::Table = InboxSpec::new("firmware_updates", Column::new("id").generated())
110 /// .advisory(&[KeyPart::Literal("firmware-"), KeyPart::Column("device")])
111 /// .payload(Column::new("payload"));
112 ///
113 /// fn id(&self) -> &i64 {
114 /// &self.id
115 /// }
116 /// }
117 /// # impl PayloadRow for FirmwareUpdate {
118 /// # type Column = Vec<u8>;
119 /// # fn payload(&self) -> &[u8] { &self.payload }
120 /// # }
121 /// # #[derive(Deserialize)]
122 /// # struct Image { version: String }
123 /// # #[subscriber(InboxQueue::<FirmwareUpdate>::new("firmware"))]
124 /// # async fn flash(image: &Image) -> HandlerOutcome {
125 /// # tracing::info!(version = %image.version, "flashing");
126 /// # HandlerOutcome::ack()
127 /// # }
128 /// # pub fn app(pool: MySqlPool) -> RustStream {
129 /// # RustStream::new(AppInfo::new("devices", "1.0.0"))
130 /// # .with_broker(SqlxBroker::<MySql>::new(pool), |b| {
131 /// # b.include(flash);
132 /// # })
133 /// # }
134 /// # }
135 /// # fn main() {}
136 /// ```
137 Advisory, Form
138);
139setting!(
140 /// Payload mode: set by [`InboxSpec::payload`](crate::InboxSpec::payload).
141 ///
142 /// # Examples
143 ///
144 /// ```
145 /// # #[cfg(feature = "postgres")]
146 /// # mod demo {
147 /// use ruststream_sqlx::dialect::Column;
148 /// use ruststream_sqlx::spec::Payload;
149 /// use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};
150 /// # use ruststream_sqlx::prelude::*;
151 /// # use serde::Deserialize;
152 /// # use sqlx::PgPool;
153 ///
154 /// #[derive(sqlx::FromRow)]
155 /// pub struct OrderEvent {
156 /// id: i64,
157 /// body: Vec<u8>,
158 /// }
159 ///
160 /// impl InboxTable for OrderEvent {
161 /// type Id = i64;
162 /// // The handler takes the message decoded from `body`, not the row.
163 /// type Table = InboxSpec<(Payload,)>;
164 /// const TABLE: Self::Table = InboxSpec::new("order_events", Column::new("id").generated())
165 /// .payload(Column::new("body"));
166 ///
167 /// fn id(&self) -> &i64 {
168 /// &self.id
169 /// }
170 /// }
171 ///
172 /// impl PayloadRow for OrderEvent {
173 /// type Column = Vec<u8>;
174 ///
175 /// fn payload(&self) -> &[u8] {
176 /// &self.body
177 /// }
178 /// }
179 /// # #[derive(Deserialize)]
180 /// # struct Placed { order: i64 }
181 /// # #[subscriber(InboxQueue::<OrderEvent>::new("orders"))]
182 /// # async fn placed(event: &Placed) -> HandlerOutcome {
183 /// # tracing::info!(order = event.order, "placed");
184 /// # HandlerOutcome::ack()
185 /// # }
186 /// # pub fn app(pool: PgPool) -> RustStream {
187 /// # RustStream::new(AppInfo::new("orders", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
188 /// # b.include(placed);
189 /// # })
190 /// # }
191 /// # }
192 /// # fn main() {}
193 /// ```
194 Payload, Message
195);
196setting!(
197 /// The partition key: set by [`InboxSpec::partition_key`](crate::InboxSpec::partition_key).
198 ///
199 /// # Examples
200 ///
201 /// ```
202 /// # #[cfg(feature = "postgres")]
203 /// # mod demo {
204 /// use ruststream_sqlx::dialect::Column;
205 /// use ruststream_sqlx::spec::{Key, Payload};
206 /// use ruststream_sqlx::{InboxSpec, InboxTable, KeyRow};
207 /// # use ruststream_sqlx::PayloadRow;
208 /// # use ruststream_sqlx::prelude::*;
209 /// # use serde::Deserialize;
210 /// # use sqlx::PgPool;
211 ///
212 /// #[derive(sqlx::FromRow)]
213 /// pub struct MeterReading {
214 /// id: i64,
215 /// meter: Option<String>,
216 /// payload: Vec<u8>,
217 /// }
218 ///
219 /// impl InboxTable for MeterReading {
220 /// type Id = i64;
221 /// // One meter's readings keep their order; a reading without a meter has no key.
222 /// type Table = InboxSpec<(Key, Payload)>;
223 /// const TABLE: Self::Table = InboxSpec::new("meter_readings", Column::new("id").generated())
224 /// .partition_key(Column::new("meter"))
225 /// .payload(Column::new("payload"));
226 ///
227 /// fn id(&self) -> &i64 {
228 /// &self.id
229 /// }
230 /// }
231 ///
232 /// impl KeyRow for MeterReading {
233 /// type Key = Option<String>;
234 ///
235 /// fn partition_key(&self) -> &Option<String> {
236 /// &self.meter
237 /// }
238 /// }
239 /// # impl PayloadRow for MeterReading {
240 /// # type Column = Vec<u8>;
241 /// # fn payload(&self) -> &[u8] { &self.payload }
242 /// # }
243 /// # #[derive(Deserialize)]
244 /// # struct Reading { kwh: f64 }
245 /// # #[subscriber(InboxQueue::<MeterReading>::new("readings"))]
246 /// # async fn bill(reading: &Reading) -> HandlerOutcome {
247 /// # tracing::info!(kwh = reading.kwh, "billing");
248 /// # HandlerOutcome::ack()
249 /// # }
250 /// # pub fn app(pool: PgPool) -> RustStream {
251 /// # RustStream::new(AppInfo::new("metering", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
252 /// # b.include(bill);
253 /// # })
254 /// # }
255 /// # }
256 /// # fn main() {}
257 /// ```
258 Key, Key
259);
260setting!(
261 /// The attempt count: set by [`InboxSpec::attempt`](crate::InboxSpec::attempt).
262 ///
263 /// # Examples
264 ///
265 /// ```
266 /// # #[cfg(feature = "mysql")]
267 /// # mod demo {
268 /// use ruststream_sqlx::dialect::Column;
269 /// use ruststream_sqlx::spec::{Attempt, Payload};
270 /// use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable};
271 /// # use ruststream_sqlx::PayloadRow;
272 /// # use ruststream_sqlx::prelude::*;
273 /// # use serde::Deserialize;
274 /// # use sqlx::{MySql, MySqlPool};
275 ///
276 /// #[derive(sqlx::FromRow)]
277 /// pub struct CallbackJob {
278 /// id: i64,
279 /// attempt: i16,
280 /// payload: Vec<u8>,
281 /// }
282 ///
283 /// impl InboxTable for CallbackJob {
284 /// type Id = i64;
285 /// // `attempt` counts the tries, and the delivery reports it.
286 /// type Table = InboxSpec<(Attempt, Payload)>;
287 /// const TABLE: Self::Table = InboxSpec::new("callbacks", Column::new("id").generated())
288 /// .attempt(Column::new("attempt").generated())
289 /// .payload(Column::new("payload"));
290 ///
291 /// fn id(&self) -> &i64 {
292 /// &self.id
293 /// }
294 /// }
295 ///
296 /// impl AttemptRow for CallbackJob {
297 /// type Attempt = i16;
298 ///
299 /// fn attempt(&self) -> &i16 {
300 /// &self.attempt
301 /// }
302 /// }
303 /// # impl PayloadRow for CallbackJob {
304 /// # type Column = Vec<u8>;
305 /// # fn payload(&self) -> &[u8] { &self.payload }
306 /// # }
307 /// # #[derive(Deserialize)]
308 /// # struct Callback { url: String }
309 /// # #[subscriber(InboxQueue::<CallbackJob>::new("callbacks"))]
310 /// # async fn call(callback: &Callback, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
311 /// # tracing::info!(url = %callback.url, ?attempt, "calling back");
312 /// # HandlerOutcome::ack()
313 /// # }
314 /// # pub fn app(pool: MySqlPool) -> RustStream {
315 /// # RustStream::new(AppInfo::new("callbacks", "1.0.0"))
316 /// # .with_broker(SqlxBroker::<MySql>::new(pool), |b| {
317 /// # b.include(call);
318 /// # })
319 /// # }
320 /// # }
321 /// # fn main() {}
322 /// ```
323 Attempt, Attempt
324);
325setting!(
326 /// The attempt count of a field sqlx converts from the column's `Decoded`
327 /// (`#[sqlx(try_from = "..")]`): a row that does not decode reads its attempt as `Decoded`.
328 /// Set by `#[derive(Inbox)]`.
329 #[doc(hidden)]
330 AttemptFrom<Decoded>, Attempt
331);
332setting!(
333 /// The delivery's headers, read from one column: set by
334 /// [`InboxSpec::headers`](crate::InboxSpec::headers).
335 ///
336 /// # Examples
337 ///
338 /// ```
339 /// # #[cfg(all(feature = "sqlite", feature = "chrono", feature = "json"))]
340 /// # mod demo {
341 /// use std::collections::BTreeMap;
342 ///
343 /// use chrono::{DateTime, Utc};
344 /// use ruststream_sqlx::dialect::Column;
345 /// use ruststream_sqlx::spec::{Headers, Lease, Payload};
346 /// use ruststream_sqlx::{HeaderRow, InboxSpec, InboxTable};
347 /// use sqlx::types::Json;
348 /// # use ruststream_sqlx::PayloadRow;
349 /// # use ruststream_sqlx::prelude::*;
350 /// # use serde::Deserialize;
351 /// # use sqlx::{Sqlite, SqlitePool};
352 ///
353 /// #[derive(sqlx::FromRow)]
354 /// pub struct Relay {
355 /// id: i64,
356 /// headers: Json<BTreeMap<String, String>>,
357 /// payload: Vec<u8>,
358 /// }
359 ///
360 /// impl InboxTable for Relay {
361 /// type Id = i64;
362 /// // The headers a relayed message carried come back from the `headers` JSON column.
363 /// type Table = InboxSpec<(Lease<DateTime<Utc>>, Headers, Payload)>;
364 /// const TABLE: Self::Table = InboxSpec::new("relay", Column::new("id").generated())
365 /// .lease(Column::new("locked_until"))
366 /// .headers(Column::new("headers"))
367 /// .payload(Column::new("payload"));
368 ///
369 /// fn id(&self) -> &i64 {
370 /// &self.id
371 /// }
372 /// }
373 ///
374 /// impl HeaderRow for Relay {
375 /// type Column = Json<BTreeMap<String, String>>;
376 ///
377 /// fn headers_mut(&mut self) -> &mut Self::Column {
378 /// &mut self.headers
379 /// }
380 /// }
381 /// # impl PayloadRow for Relay {
382 /// # type Column = Vec<u8>;
383 /// # fn payload(&self) -> &[u8] { &self.payload }
384 /// # }
385 /// # #[derive(Deserialize)]
386 /// # struct Forwarded { id: String }
387 /// # #[subscriber(InboxQueue::<Relay>::new("relay"))]
388 /// # async fn forward(message: &Forwarded, ctx: &mut Context<'_>) -> HandlerOutcome {
389 /// # let source = ctx.headers().get_str("source").unwrap_or("unknown");
390 /// # tracing::info!(id = %message.id, source, "forwarding");
391 /// # HandlerOutcome::ack()
392 /// # }
393 /// # pub fn app(pool: SqlitePool) -> RustStream {
394 /// # RustStream::new(AppInfo::new("relay", "1.0.0"))
395 /// # .with_broker(SqlxBroker::<Sqlite>::new(pool), |b| {
396 /// # b.include(forward);
397 /// # })
398 /// # }
399 /// # }
400 /// # fn main() {}
401 /// ```
402 Headers, Headers
403);
404setting!(
405 /// The delivery's headers, built from the row's header fields: set by
406 /// [`InboxSpec::header_fields`](crate::InboxSpec::header_fields).
407 ///
408 /// # Examples
409 ///
410 /// ```
411 /// # #[cfg(feature = "postgres")]
412 /// # mod demo {
413 /// use ruststream::HeaderMap;
414 /// use ruststream::runtime::{Input, SoloCarried};
415 /// use ruststream_sqlx::dialect::Column;
416 /// use ruststream_sqlx::spec;
417 /// use ruststream_sqlx::{HeaderFields, InboxSpec, InboxTable, put_header};
418 /// # use ruststream_sqlx::prelude::*;
419 /// # use sqlx::PgPool;
420 ///
421 /// #[derive(Debug, Clone, sqlx::FromRow)]
422 /// pub struct Signal {
423 /// id: i64,
424 /// trace: Option<String>,
425 /// sensor: String,
426 /// value: f64,
427 /// }
428 ///
429 /// impl InboxTable for Signal {
430 /// type Id = i64;
431 /// // `trace` reads as a header; a row without one has no `trace` header.
432 /// type Table = InboxSpec<(spec::HeaderFields,)>;
433 /// const TABLE: Self::Table = InboxSpec::new("signals", Column::new("id").generated())
434 /// .data(&[Column::new("trace"), Column::new("sensor"), Column::new("value")])
435 /// .header_fields();
436 ///
437 /// fn id(&self) -> &i64 {
438 /// &self.id
439 /// }
440 /// }
441 ///
442 /// impl HeaderFields for Signal {
443 /// const NAMES: &'static [&'static str] = &["trace"];
444 ///
445 /// fn header_map(&self) -> HeaderMap {
446 /// let mut headers = HeaderMap::with_capacity(Self::NAMES.len());
447 /// put_header(&mut headers, "trace", &self.trace);
448 /// headers
449 /// }
450 /// }
451 /// # impl Input for Signal {
452 /// # type Axis = SoloCarried<Self>;
453 /// # }
454 /// # #[subscriber(InboxQueue::<Signal>::new("signals"))]
455 /// # async fn record(signal: &Signal) -> HandlerOutcome {
456 /// # tracing::info!(sensor = %signal.sensor, value = signal.value, "recorded");
457 /// # HandlerOutcome::ack()
458 /// # }
459 /// # pub fn app(pool: PgPool) -> RustStream {
460 /// # RustStream::new(AppInfo::new("sensors", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
461 /// # b.include(record);
462 /// # })
463 /// # }
464 /// # }
465 /// # fn main() {}
466 /// ```
467 HeaderFields, Headers
468);
469setting!(
470 /// The time a retried row is due, in `Time`: set by
471 /// [`InboxSpec::retry_after`](crate::InboxSpec::retry_after).
472 ///
473 /// # Examples
474 ///
475 /// ```
476 /// # #[cfg(all(feature = "mysql", feature = "chrono"))]
477 /// # mod demo {
478 /// use chrono::{DateTime, Utc};
479 /// use ruststream_sqlx::dialect::Column;
480 /// use ruststream_sqlx::spec::{Payload, RetryAfter};
481 /// use ruststream_sqlx::{InboxSpec, InboxTable};
482 /// # use std::time::Duration;
483 /// # use ruststream_sqlx::PayloadRow;
484 /// # use ruststream_sqlx::prelude::*;
485 /// # use serde::Deserialize;
486 /// # use sqlx::{MySql, MySqlPool};
487 ///
488 /// #[derive(sqlx::FromRow)]
489 /// pub struct Delivery {
490 /// id: i64,
491 /// payload: Vec<u8>,
492 /// }
493 ///
494 /// impl InboxTable for Delivery {
495 /// type Id = i64;
496 /// // A retried webhook waits until `retry_after`, a `DATETIME(6)` read as UTC.
497 /// type Table = InboxSpec<(RetryAfter<DateTime<Utc>>, Payload)>;
498 /// const TABLE: Self::Table = InboxSpec::new("webhook_deliveries", Column::new("id").generated())
499 /// .retry_after(Column::new("retry_after").generated())
500 /// .payload(Column::new("payload"));
501 ///
502 /// fn id(&self) -> &i64 {
503 /// &self.id
504 /// }
505 /// }
506 /// # impl PayloadRow for Delivery {
507 /// # type Column = Vec<u8>;
508 /// # fn payload(&self) -> &[u8] { &self.payload }
509 /// # }
510 /// # #[derive(Deserialize)]
511 /// # struct Hook { url: String }
512 /// # #[subscriber(InboxQueue::<Delivery>::new("webhooks"))]
513 /// # async fn deliver(hook: &Hook) -> HandlerOutcome {
514 /// # tracing::info!(url = %hook.url, "delivering");
515 /// # HandlerOutcome::retry_after(Duration::from_secs(300))
516 /// # }
517 /// # pub fn app(pool: MySqlPool) -> RustStream {
518 /// # RustStream::new(AppInfo::new("webhooks", "1.0.0"))
519 /// # .with_broker(SqlxBroker::<MySql>::new(pool), |b| {
520 /// # b.include(deliver);
521 /// # })
522 /// # }
523 /// # }
524 /// # fn main() {}
525 /// ```
526 RetryAfter<Time>, RetryAfter
527);
528setting!(
529 /// The time a row was processed, in `Time`: set by
530 /// [`InboxSpec::processed_at`](crate::InboxSpec::processed_at).
531 ///
532 /// # Examples
533 ///
534 /// ```
535 /// # #[cfg(all(feature = "postgres", feature = "chrono"))]
536 /// # mod demo {
537 /// use chrono::{DateTime, Utc};
538 /// use ruststream_sqlx::dialect::Column;
539 /// use ruststream_sqlx::spec::{Payload, ProcessedAt};
540 /// use ruststream_sqlx::{InboxSpec, InboxTable};
541 /// # use ruststream_sqlx::PayloadRow;
542 /// # use ruststream_sqlx::prelude::*;
543 /// # use serde::Deserialize;
544 /// # use sqlx::PgPool;
545 ///
546 /// #[derive(sqlx::FromRow)]
547 /// pub struct Import {
548 /// id: i64,
549 /// payload: Vec<u8>,
550 /// }
551 ///
552 /// impl InboxTable for Import {
553 /// type Id = i64;
554 /// // A finished import keeps its row, with the time it finished in `done_at`.
555 /// type Table = InboxSpec<(ProcessedAt<DateTime<Utc>>, Payload)>;
556 /// const TABLE: Self::Table = InboxSpec::new("imports", Column::new("id").generated())
557 /// .processed_at(Column::new("done_at").generated())
558 /// .payload(Column::new("payload"));
559 ///
560 /// fn id(&self) -> &i64 {
561 /// &self.id
562 /// }
563 /// }
564 /// # impl PayloadRow for Import {
565 /// # type Column = Vec<u8>;
566 /// # fn payload(&self) -> &[u8] { &self.payload }
567 /// # }
568 /// # #[derive(Deserialize)]
569 /// # struct Csv { file: String }
570 /// # #[subscriber(InboxQueue::<Import>::new("imports"))]
571 /// # async fn import(csv: &Csv) -> HandlerOutcome {
572 /// # tracing::info!(file = %csv.file, "importing");
573 /// # HandlerOutcome::ack()
574 /// # }
575 /// # pub fn app(pool: PgPool) -> RustStream {
576 /// # RustStream::new(AppInfo::new("imports", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
577 /// # b.include(import);
578 /// # })
579 /// # }
580 /// # }
581 /// # fn main() {}
582 /// ```
583 ProcessedAt<Time>, ProcessedAt
584);
585setting!(
586 /// Groups kept in order: set by [`InboxSpec::fifo_group`](crate::InboxSpec::fifo_group).
587 ///
588 /// # Examples
589 ///
590 /// ```
591 /// # #[cfg(all(feature = "postgres", feature = "chrono"))]
592 /// # mod demo {
593 /// use chrono::{DateTime, Utc};
594 /// use ruststream_sqlx::dialect::Column;
595 /// use ruststream_sqlx::spec::{Fifo, Lease, Payload};
596 /// use ruststream_sqlx::{InboxSpec, InboxTable};
597 /// # use ruststream_sqlx::PayloadRow;
598 /// # use ruststream_sqlx::prelude::*;
599 /// # use serde::Deserialize;
600 /// # use sqlx::PgPool;
601 ///
602 /// #[derive(sqlx::FromRow)]
603 /// pub struct Migration {
604 /// id: i64,
605 /// payload: Vec<u8>,
606 /// }
607 ///
608 /// impl InboxTable for Migration {
609 /// type Id = i64;
610 /// // A tenant's schema migrations run one at a time, in the order they were queued.
611 /// type Table = InboxSpec<(Lease<DateTime<Utc>>, Fifo, Payload)>;
612 /// const TABLE: Self::Table = InboxSpec::new("migrations", Column::new("id").generated())
613 /// .lease(Column::new("locked_until"))
614 /// .fifo_group(Column::new("tenant"))
615 /// .payload(Column::new("payload"));
616 ///
617 /// fn id(&self) -> &i64 {
618 /// &self.id
619 /// }
620 /// }
621 /// # impl PayloadRow for Migration {
622 /// # type Column = Vec<u8>;
623 /// # fn payload(&self) -> &[u8] { &self.payload }
624 /// # }
625 /// # #[derive(Deserialize)]
626 /// # struct Step { version: u32 }
627 /// # #[subscriber(InboxQueue::<Migration>::new("acme"))]
628 /// # async fn migrate(step: &Step) -> HandlerOutcome {
629 /// # tracing::info!(version = step.version, "migrating acme");
630 /// # HandlerOutcome::ack()
631 /// # }
632 /// # pub fn app(pool: PgPool) -> RustStream {
633 /// # RustStream::new(AppInfo::new("tenants", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
634 /// # b.include(migrate);
635 /// # })
636 /// # }
637 /// # }
638 /// # fn main() {}
639 /// ```
640 Fifo, Fifo
641);
642setting!(
643 /// The clock the table reads now from, a [`TimeSource`](crate::TimeSource): set by
644 /// [`InboxSpec::clock`](crate::InboxSpec::clock).
645 ///
646 /// # Examples
647 ///
648 /// ```
649 /// # #[cfg(all(feature = "mysql", feature = "chrono"))]
650 /// # mod demo {
651 /// use std::time::{Duration, SystemTime, UNIX_EPOCH};
652 ///
653 /// use chrono::{DateTime, Utc};
654 /// use ruststream_sqlx::dialect::Column;
655 /// use ruststream_sqlx::spec::{self, Payload, RetryAfter};
656 /// use ruststream_sqlx::{Clock, InboxSpec, InboxTable};
657 /// # use ruststream_sqlx::PayloadRow;
658 /// # use ruststream_sqlx::prelude::*;
659 /// # use serde::Deserialize;
660 /// # use sqlx::{MySql, MySqlPool};
661 ///
662 /// /// The host's time in whole seconds, as the table's `DATETIME(0)` columns keep it.
663 /// pub struct WholeSeconds;
664 ///
665 /// impl Clock for WholeSeconds {
666 /// fn now() -> SystemTime {
667 /// let since = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default();
668 /// UNIX_EPOCH + Duration::from_secs(since.as_secs())
669 /// }
670 /// }
671 ///
672 /// #[derive(sqlx::FromRow)]
673 /// pub struct Digest {
674 /// id: i64,
675 /// payload: Vec<u8>,
676 /// }
677 ///
678 /// impl InboxTable for Digest {
679 /// type Id = i64;
680 /// // Due times are bound from `WholeSeconds`, so they compare equal to what the column stores.
681 /// type Table = InboxSpec<(spec::Clock<WholeSeconds>, RetryAfter<DateTime<Utc>>, Payload)>;
682 /// const TABLE: Self::Table = InboxSpec::new("digests", Column::new("id").generated())
683 /// .clock::<WholeSeconds>()
684 /// .retry_after(Column::new("retry_after").generated())
685 /// .payload(Column::new("payload"));
686 ///
687 /// fn id(&self) -> &i64 {
688 /// &self.id
689 /// }
690 /// }
691 /// # impl PayloadRow for Digest {
692 /// # type Column = Vec<u8>;
693 /// # fn payload(&self) -> &[u8] { &self.payload }
694 /// # }
695 /// # #[derive(Deserialize)]
696 /// # struct Daily { user: String }
697 /// # #[subscriber(InboxQueue::<Digest>::new("digests"))]
698 /// # async fn digest(daily: &Daily) -> HandlerOutcome {
699 /// # tracing::info!(user = %daily.user, "sending a digest");
700 /// # HandlerOutcome::retry_after(Duration::from_secs(86_400))
701 /// # }
702 /// # pub fn app(pool: MySqlPool) -> RustStream {
703 /// # RustStream::new(AppInfo::new("digests", "1.0.0"))
704 /// # .with_broker(SqlxBroker::<MySql>::new(pool), |b| {
705 /// # b.include(digest);
706 /// # })
707 /// # }
708 /// # }
709 /// # fn main() {}
710 /// ```
711 Clock<Source>, Clock
712);
713setting!(
714 /// What the table's transactions open at, a [`level`] marker: set by
715 /// [`InboxSpec::opens`](crate::InboxSpec::opens).
716 ///
717 /// # Examples
718 ///
719 /// ```
720 /// # #[cfg(all(feature = "sqlite", feature = "chrono"))]
721 /// # mod demo {
722 /// use chrono::{DateTime, Utc};
723 /// use ruststream_sqlx::dialect::{Column, level};
724 /// use ruststream_sqlx::spec::{Lease, Opens, Payload};
725 /// use ruststream_sqlx::{InboxSpec, InboxTable};
726 /// # use ruststream_sqlx::PayloadRow;
727 /// # use ruststream_sqlx::prelude::*;
728 /// # use serde::Deserialize;
729 /// # use sqlx::{Sqlite, SqlitePool};
730 ///
731 /// #[derive(sqlx::FromRow)]
732 /// pub struct PrintJob {
733 /// id: i64,
734 /// payload: Vec<u8>,
735 /// }
736 ///
737 /// impl InboxTable for PrintJob {
738 /// type Id = i64;
739 /// // A transaction takes the database's write lock as it begins, not at its first write.
740 /// type Table = InboxSpec<(Lease<DateTime<Utc>>, Opens<level::Immediate>, Payload)>;
741 /// const TABLE: Self::Table = InboxSpec::new("print_jobs", Column::new("id").generated())
742 /// .lease(Column::new("locked_until"))
743 /// .opens::<level::Immediate>()
744 /// .payload(Column::new("payload"));
745 ///
746 /// fn id(&self) -> &i64 {
747 /// &self.id
748 /// }
749 /// }
750 /// # impl PayloadRow for PrintJob {
751 /// # type Column = Vec<u8>;
752 /// # fn payload(&self) -> &[u8] { &self.payload }
753 /// # }
754 /// # #[derive(Deserialize)]
755 /// # struct Document { pages: u32 }
756 /// # #[subscriber(InboxQueue::<PrintJob>::new("printer"))]
757 /// # async fn print(document: &Document) -> HandlerOutcome {
758 /// # tracing::info!(pages = document.pages, "printing");
759 /// # HandlerOutcome::ack()
760 /// # }
761 /// # pub fn app(pool: SqlitePool) -> RustStream {
762 /// # RustStream::new(AppInfo::new("printing", "1.0.0"))
763 /// # .with_broker(SqlxBroker::<Sqlite>::new(pool), |b| {
764 /// # b.include(print);
765 /// # })
766 /// # }
767 /// # }
768 /// # fn main() {}
769 /// ```
770 Opens<Level>, Opening
771);
772
773/// The events a service writes itself, each set by [`InboxSpec::own`](crate::InboxSpec::own); the
774/// row implements the event's trait.
775pub mod own {
776 use super::declaration::{Declaration, Set, Unset, setting};
777
778 setting!(
779 /// The service's own [`Claim`](crate::Claim): set by
780 /// [`InboxSpec::own`](crate::InboxSpec::own).
781 ///
782 /// # Examples
783 ///
784 /// ```
785 /// # #[cfg(feature = "postgres")]
786 /// # mod demo {
787 /// use ruststream_sqlx::dialect::Column;
788 /// use ruststream_sqlx::spec::{Payload, own};
789 /// use ruststream_sqlx::{Claim, InboxSpec, InboxTable};
790 /// use sqlx::{PgConnection, Postgres};
791 /// # use ruststream_sqlx::PayloadRow;
792 /// # use ruststream_sqlx::prelude::*;
793 /// # use serde::Deserialize;
794 /// # use sqlx::PgPool;
795 ///
796 /// #[derive(sqlx::FromRow)]
797 /// pub struct Callout {
798 /// id: i64,
799 /// payload: Vec<u8>,
800 /// }
801 ///
802 /// impl InboxTable for Callout {
803 /// type Id = i64;
804 /// type Table = InboxSpec<(Payload, own::Claim)>;
805 /// const TABLE: Self::Table = InboxSpec::new("callouts", Column::new("id").generated())
806 /// .group(Column::new("name"))
807 /// .payload(Column::new("payload"))
808 /// .own::<own::Claim>();
809 ///
810 /// fn id(&self) -> &i64 {
811 /// &self.id
812 /// }
813 /// }
814 ///
815 /// // The callouts of paying customers are claimed before the others.
816 /// impl Claim<Postgres> for Callout {
817 /// async fn claim(
818 /// conn: &mut PgConnection,
819 /// queue: &str,
820 /// limit: i64,
821 /// ) -> Result<Vec<i64>, sqlx::Error> {
822 /// sqlx::query_scalar(
823 /// "SELECT c.id FROM callouts c JOIN customers u ON u.id = c.customer \
824 /// WHERE c.name = $1 ORDER BY u.paying DESC, c.id LIMIT $2 FOR UPDATE OF c SKIP LOCKED",
825 /// )
826 /// .bind(queue)
827 /// .bind(limit)
828 /// .fetch_all(conn)
829 /// .await
830 /// }
831 /// }
832 /// # impl PayloadRow for Callout {
833 /// # type Column = Vec<u8>;
834 /// # fn payload(&self) -> &[u8] { &self.payload }
835 /// # }
836 /// # #[derive(Deserialize)]
837 /// # struct Call { phone: String }
838 /// # #[subscriber(InboxQueue::<Callout>::new("callouts"))]
839 /// # async fn dial(call: &Call) -> HandlerOutcome {
840 /// # tracing::info!(phone = %call.phone, "dialing");
841 /// # HandlerOutcome::ack()
842 /// # }
843 /// # pub fn app(pool: PgPool) -> RustStream {
844 /// # RustStream::new(AppInfo::new("callouts", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
845 /// # b.include(dial);
846 /// # })
847 /// # }
848 /// # }
849 /// # fn main() {}
850 /// ```
851 Claim, OwnClaim
852 );
853 setting!(
854 /// The service's own [`Fetch`](crate::Fetch): set by
855 /// [`InboxSpec::own`](crate::InboxSpec::own).
856 ///
857 /// # Examples
858 ///
859 /// ```
860 /// # #[cfg(feature = "postgres")]
861 /// # mod demo {
862 /// use ruststream_sqlx::dialect::Column;
863 /// use ruststream_sqlx::spec::{Payload, own};
864 /// use ruststream_sqlx::{Fetch, InboxSpec, InboxTable};
865 /// use sqlx::{PgConnection, Postgres};
866 /// # use ruststream_sqlx::PayloadRow;
867 /// # use ruststream_sqlx::prelude::*;
868 /// # use serde::Deserialize;
869 /// # use sqlx::PgPool;
870 ///
871 /// #[derive(sqlx::FromRow)]
872 /// pub struct RenderJob {
873 /// id: i64,
874 /// payload: Vec<u8>,
875 /// }
876 ///
877 /// impl InboxTable for RenderJob {
878 /// type Id = i64;
879 /// type Table = InboxSpec<(Payload, own::Fetch)>;
880 /// const TABLE: Self::Table = InboxSpec::new("render_jobs", Column::new("id").generated())
881 /// .payload(Column::new("payload"))
882 /// .own::<own::Fetch>();
883 ///
884 /// fn id(&self) -> &i64 {
885 /// &self.id
886 /// }
887 /// }
888 ///
889 /// // The job holds a document's id, and its message is that document's latest body.
890 /// impl Fetch<Postgres> for RenderJob {
891 /// async fn fetch(conn: &mut PgConnection, ids: &[i64]) -> Result<Vec<Self>, sqlx::Error> {
892 /// sqlx::query_as(
893 /// "SELECT j.id, d.body AS payload FROM render_jobs j \
894 /// JOIN documents d ON d.id = j.document WHERE j.id = ANY($1)",
895 /// )
896 /// .bind(ids)
897 /// .fetch_all(conn)
898 /// .await
899 /// }
900 /// }
901 /// # impl PayloadRow for RenderJob {
902 /// # type Column = Vec<u8>;
903 /// # fn payload(&self) -> &[u8] { &self.payload }
904 /// # }
905 /// # #[derive(Deserialize)]
906 /// # struct Document { title: String }
907 /// # #[subscriber(InboxQueue::<RenderJob>::new("renders"))]
908 /// # async fn render(document: &Document) -> HandlerOutcome {
909 /// # tracing::info!(title = %document.title, "rendering");
910 /// # HandlerOutcome::ack()
911 /// # }
912 /// # pub fn app(pool: PgPool) -> RustStream {
913 /// # RustStream::new(AppInfo::new("documents", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
914 /// # b.include(render);
915 /// # })
916 /// # }
917 /// # }
918 /// # fn main() {}
919 /// ```
920 Fetch, OwnFetch
921 );
922 setting!(
923 /// The service's own [`Ack`](crate::Ack): set by
924 /// [`InboxSpec::own`](crate::InboxSpec::own).
925 ///
926 /// # Examples
927 ///
928 /// ```
929 /// # #[cfg(feature = "postgres")]
930 /// # mod demo {
931 /// use ruststream_sqlx::dialect::Column;
932 /// use ruststream_sqlx::spec::{Payload, own};
933 /// use ruststream_sqlx::{Ack, InboxSpec, InboxTable};
934 /// use sqlx::{PgConnection, Postgres};
935 /// # use ruststream_sqlx::PayloadRow;
936 /// # use ruststream_sqlx::prelude::*;
937 /// # use serde::Deserialize;
938 /// # use sqlx::PgPool;
939 ///
940 /// #[derive(sqlx::FromRow)]
941 /// pub struct UploadJob {
942 /// id: i64,
943 /// payload: Vec<u8>,
944 /// }
945 ///
946 /// impl InboxTable for UploadJob {
947 /// type Id = i64;
948 /// type Table = InboxSpec<(Payload, own::Ack)>;
949 /// const TABLE: Self::Table = InboxSpec::new("uploads", Column::new("id").generated())
950 /// .payload(Column::new("payload"))
951 /// .own::<own::Ack>();
952 ///
953 /// fn id(&self) -> &i64 {
954 /// &self.id
955 /// }
956 /// }
957 ///
958 /// // A finished upload leaves a line in `upload_log` as its row goes.
959 /// impl Ack<Postgres> for UploadJob {
960 /// async fn ack(conn: &mut PgConnection, id: &i64) -> Result<(), sqlx::Error> {
961 /// sqlx::query("INSERT INTO upload_log (upload, finished_at) VALUES ($1, now())")
962 /// .bind(id)
963 /// .execute(&mut *conn)
964 /// .await?;
965 /// sqlx::query("DELETE FROM uploads WHERE id = $1")
966 /// .bind(id)
967 /// .execute(conn)
968 /// .await?;
969 /// Ok(())
970 /// }
971 /// }
972 /// # impl PayloadRow for UploadJob {
973 /// # type Column = Vec<u8>;
974 /// # fn payload(&self) -> &[u8] { &self.payload }
975 /// # }
976 /// # #[derive(Deserialize)]
977 /// # struct Upload { key: String }
978 /// # #[subscriber(InboxQueue::<UploadJob>::new("uploads"))]
979 /// # async fn store(upload: &Upload) -> HandlerOutcome {
980 /// # tracing::info!(key = %upload.key, "storing");
981 /// # HandlerOutcome::ack()
982 /// # }
983 /// # pub fn app(pool: PgPool) -> RustStream {
984 /// # RustStream::new(AppInfo::new("uploads", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
985 /// # b.include(store);
986 /// # })
987 /// # }
988 /// # }
989 /// # fn main() {}
990 /// ```
991 Ack, OwnAck
992 );
993 setting!(
994 /// The service's own [`Retry`](crate::Retry): set by
995 /// [`InboxSpec::own`](crate::InboxSpec::own).
996 ///
997 /// # Examples
998 ///
999 /// ```
1000 /// # #[cfg(feature = "postgres")]
1001 /// # mod demo {
1002 /// use ruststream_sqlx::dialect::Column;
1003 /// use ruststream_sqlx::spec::{Payload, own};
1004 /// use ruststream_sqlx::{InboxSpec, InboxTable, Retry};
1005 /// use sqlx::{PgConnection, Postgres};
1006 /// # use ruststream_sqlx::PayloadRow;
1007 /// # use ruststream_sqlx::prelude::*;
1008 /// # use serde::Deserialize;
1009 /// # use sqlx::PgPool;
1010 ///
1011 /// #[derive(sqlx::FromRow)]
1012 /// pub struct CrawlJob {
1013 /// id: i64,
1014 /// payload: Vec<u8>,
1015 /// }
1016 ///
1017 /// impl InboxTable for CrawlJob {
1018 /// type Id = i64;
1019 /// type Table = InboxSpec<(Payload, own::Retry)>;
1020 /// const TABLE: Self::Table = InboxSpec::new("crawl_jobs", Column::new("id").generated())
1021 /// .payload(Column::new("payload"))
1022 /// .own::<own::Retry>();
1023 ///
1024 /// fn id(&self) -> &i64 {
1025 /// &self.id
1026 /// }
1027 /// }
1028 ///
1029 /// // A retry records when the page last failed, for the crawler's dashboard.
1030 /// impl Retry<Postgres> for CrawlJob {
1031 /// async fn retry(conn: &mut PgConnection, id: &i64) -> Result<(), sqlx::Error> {
1032 /// sqlx::query("UPDATE crawl_jobs SET last_failure = now() WHERE id = $1")
1033 /// .bind(id)
1034 /// .execute(conn)
1035 /// .await?;
1036 /// Ok(())
1037 /// }
1038 /// }
1039 /// # impl PayloadRow for CrawlJob {
1040 /// # type Column = Vec<u8>;
1041 /// # fn payload(&self) -> &[u8] { &self.payload }
1042 /// # }
1043 /// # #[derive(Deserialize)]
1044 /// # struct Page { url: String }
1045 /// # #[subscriber(InboxQueue::<CrawlJob>::new("crawl"))]
1046 /// # async fn crawl(page: &Page) -> HandlerOutcome {
1047 /// # tracing::info!(url = %page.url, "crawling");
1048 /// # HandlerOutcome::retry()
1049 /// # }
1050 /// # pub fn app(pool: PgPool) -> RustStream {
1051 /// # RustStream::new(AppInfo::new("crawler", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
1052 /// # b.include(crawl);
1053 /// # })
1054 /// # }
1055 /// # }
1056 /// # fn main() {}
1057 /// ```
1058 Retry, OwnRetry
1059 );
1060 setting!(
1061 /// The service's own [`RetryAfter`](crate::RetryAfter): set by
1062 /// [`InboxSpec::own`](crate::InboxSpec::own).
1063 ///
1064 /// # Examples
1065 ///
1066 /// ```
1067 /// # #[cfg(all(feature = "postgres", feature = "chrono"))]
1068 /// # mod demo {
1069 /// use std::time::Duration;
1070 ///
1071 /// use chrono::{DateTime, Utc};
1072 /// use ruststream_sqlx::dialect::Column;
1073 /// use ruststream_sqlx::spec::{self, Payload, own};
1074 /// use ruststream_sqlx::{InboxSpec, InboxTable, RetryAfter};
1075 /// use sqlx::{PgConnection, Postgres};
1076 /// # use ruststream_sqlx::PayloadRow;
1077 /// # use ruststream_sqlx::prelude::*;
1078 /// # use serde::Deserialize;
1079 /// # use sqlx::PgPool;
1080 ///
1081 /// #[derive(sqlx::FromRow)]
1082 /// pub struct QuoteJob {
1083 /// id: i64,
1084 /// payload: Vec<u8>,
1085 /// }
1086 ///
1087 /// impl InboxTable for QuoteJob {
1088 /// type Id = i64;
1089 /// type Table = InboxSpec<(spec::RetryAfter<DateTime<Utc>>, Payload, own::RetryAfter)>;
1090 /// const TABLE: Self::Table = InboxSpec::new("quote_jobs", Column::new("id").generated())
1091 /// .retry_after(Column::new("retry_after").generated())
1092 /// .payload(Column::new("payload"))
1093 /// .own::<own::RetryAfter>();
1094 ///
1095 /// fn id(&self) -> &i64 {
1096 /// &self.id
1097 /// }
1098 /// }
1099 ///
1100 /// // The wait doubles with every delayed retry of the row, starting from the handler's delay.
1101 /// impl RetryAfter<Postgres> for QuoteJob {
1102 /// async fn retry_after(
1103 /// conn: &mut PgConnection,
1104 /// id: &i64,
1105 /// delay: Duration,
1106 /// ) -> Result<(), sqlx::Error> {
1107 /// sqlx::query(
1108 /// "UPDATE quote_jobs SET delays = delays + 1, \
1109 /// retry_after = now() + make_interval(secs => $2 * power(2, delays)) WHERE id = $1",
1110 /// )
1111 /// .bind(id)
1112 /// .bind(delay.as_secs_f64())
1113 /// .execute(conn)
1114 /// .await?;
1115 /// Ok(())
1116 /// }
1117 /// }
1118 /// # impl PayloadRow for QuoteJob {
1119 /// # type Column = Vec<u8>;
1120 /// # fn payload(&self) -> &[u8] { &self.payload }
1121 /// # }
1122 /// # #[derive(Deserialize)]
1123 /// # struct Quote { symbol: String }
1124 /// # #[subscriber(InboxQueue::<QuoteJob>::new("quotes"))]
1125 /// # async fn quote(job: &Quote) -> HandlerOutcome {
1126 /// # tracing::info!(symbol = %job.symbol, "quoting");
1127 /// # HandlerOutcome::retry_after(Duration::from_secs(10))
1128 /// # }
1129 /// # pub fn app(pool: PgPool) -> RustStream {
1130 /// # RustStream::new(AppInfo::new("quotes", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
1131 /// # b.include(quote);
1132 /// # })
1133 /// # }
1134 /// # }
1135 /// # fn main() {}
1136 /// ```
1137 RetryAfter, OwnRetryAfter
1138 );
1139 setting!(
1140 /// The service's own [`Discard`](crate::Discard): set by
1141 /// [`InboxSpec::own`](crate::InboxSpec::own).
1142 ///
1143 /// # Examples
1144 ///
1145 /// ```
1146 /// # #[cfg(feature = "postgres")]
1147 /// # mod demo {
1148 /// use ruststream_sqlx::dialect::Column;
1149 /// use ruststream_sqlx::spec::{Payload, own};
1150 /// use ruststream_sqlx::{Discard, InboxSpec, InboxTable};
1151 /// use sqlx::{PgConnection, Postgres};
1152 /// # use ruststream_sqlx::PayloadRow;
1153 /// # use ruststream_sqlx::prelude::*;
1154 /// # use serde::Deserialize;
1155 /// # use sqlx::PgPool;
1156 ///
1157 /// #[derive(sqlx::FromRow)]
1158 /// pub struct Comment {
1159 /// id: i64,
1160 /// payload: Vec<u8>,
1161 /// }
1162 ///
1163 /// impl InboxTable for Comment {
1164 /// type Id = i64;
1165 /// type Table = InboxSpec<(Payload, own::Discard)>;
1166 /// const TABLE: Self::Table = InboxSpec::new("comments", Column::new("id").generated())
1167 /// .payload(Column::new("payload"))
1168 /// .own::<own::Discard>();
1169 ///
1170 /// fn id(&self) -> &i64 {
1171 /// &self.id
1172 /// }
1173 /// }
1174 ///
1175 /// // A rejected comment moves to `rejected_comments`, stamped with the time of the rejection.
1176 /// impl Discard<Postgres> for Comment {
1177 /// async fn discard(conn: &mut PgConnection, id: &i64) -> Result<(), sqlx::Error> {
1178 /// sqlx::query(
1179 /// "WITH gone AS (DELETE FROM comments WHERE id = $1 RETURNING id, payload) \
1180 /// INSERT INTO rejected_comments (id, payload, rejected_at) \
1181 /// SELECT id, payload, now() FROM gone",
1182 /// )
1183 /// .bind(id)
1184 /// .execute(conn)
1185 /// .await?;
1186 /// Ok(())
1187 /// }
1188 /// }
1189 /// # impl PayloadRow for Comment {
1190 /// # type Column = Vec<u8>;
1191 /// # fn payload(&self) -> &[u8] { &self.payload }
1192 /// # }
1193 /// # #[derive(Deserialize)]
1194 /// # struct Text { body: String }
1195 /// # #[subscriber(InboxQueue::<Comment>::new("comments"))]
1196 /// # async fn moderate(text: &Text) -> HandlerOutcome {
1197 /// # if text.body.is_empty() { HandlerOutcome::drop() } else { HandlerOutcome::ack() }
1198 /// # }
1199 /// # pub fn app(pool: PgPool) -> RustStream {
1200 /// # RustStream::new(AppInfo::new("comments", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
1201 /// # b.include(moderate);
1202 /// # })
1203 /// # }
1204 /// # }
1205 /// # fn main() {}
1206 /// ```
1207 Discard, OwnDiscard
1208 );
1209 setting!(
1210 /// The service's own [`DeadLetter`](crate::DeadLetter): set by
1211 /// [`InboxSpec::own`](crate::InboxSpec::own).
1212 ///
1213 /// # Examples
1214 ///
1215 /// ```
1216 /// # #[cfg(feature = "postgres")]
1217 /// # mod demo {
1218 /// use ruststream_sqlx::dialect::Column;
1219 /// use ruststream_sqlx::spec::{Payload, own};
1220 /// use ruststream_sqlx::{DeadLetter, InboxSpec, InboxTable};
1221 /// use sqlx::{PgConnection, Postgres};
1222 /// # use ruststream_sqlx::PayloadRow;
1223 /// # use ruststream_sqlx::prelude::*;
1224 /// # use serde::Deserialize;
1225 /// # use sqlx::PgPool;
1226 ///
1227 /// #[derive(sqlx::FromRow)]
1228 /// pub struct Charge {
1229 /// id: i64,
1230 /// payload: Vec<u8>,
1231 /// }
1232 ///
1233 /// impl InboxTable for Charge {
1234 /// type Id = i64;
1235 /// type Table = InboxSpec<(Payload, own::DeadLetter)>;
1236 /// const TABLE: Self::Table = InboxSpec::new("charges", Column::new("id").generated())
1237 /// .payload(Column::new("payload"))
1238 /// .own::<own::DeadLetter>();
1239 ///
1240 /// fn id(&self) -> &i64 {
1241 /// &self.id
1242 /// }
1243 /// }
1244 ///
1245 /// // Every queue of the service parks its spent rows in one `dead_letters` table.
1246 /// impl DeadLetter<Postgres> for Charge {
1247 /// async fn dead_letter(
1248 /// conn: &mut PgConnection,
1249 /// id: &i64,
1250 /// destination: &str,
1251 /// ) -> Result<(), sqlx::Error> {
1252 /// sqlx::query(
1253 /// "WITH dead AS (DELETE FROM charges WHERE id = $1 RETURNING payload) \
1254 /// INSERT INTO dead_letters (source, destination, payload) \
1255 /// SELECT 'charges', $2, payload FROM dead",
1256 /// )
1257 /// .bind(id)
1258 /// .bind(destination)
1259 /// .execute(conn)
1260 /// .await?;
1261 /// Ok(())
1262 /// }
1263 /// }
1264 /// # impl PayloadRow for Charge {
1265 /// # type Column = Vec<u8>;
1266 /// # fn payload(&self) -> &[u8] { &self.payload }
1267 /// # }
1268 /// # #[derive(Deserialize)]
1269 /// # struct Card { cents: i64 }
1270 /// # #[subscriber(InboxQueue::<Charge>::new("charges"))]
1271 /// # async fn charge(card: &Card) -> HandlerOutcome {
1272 /// # tracing::info!(cents = card.cents, "charging");
1273 /// # HandlerOutcome::retry()
1274 /// # }
1275 /// # pub fn app(pool: PgPool) -> RustStream {
1276 /// # RustStream::new(AppInfo::new("charges", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
1277 /// # b.include(charge);
1278 /// # })
1279 /// # }
1280 /// # }
1281 /// # fn main() {}
1282 /// ```
1283 DeadLetter, OwnDeadLetter
1284 );
1285 setting!(
1286 /// The service's own [`Extend`](crate::Extend): set by
1287 /// [`InboxSpec::own`](crate::InboxSpec::own).
1288 ///
1289 /// # Examples
1290 ///
1291 /// ```
1292 /// # #[cfg(all(feature = "postgres", feature = "chrono"))]
1293 /// # mod demo {
1294 /// use chrono::{DateTime, Utc};
1295 /// use ruststream_sqlx::dialect::Column;
1296 /// use ruststream_sqlx::spec::{Lease, Payload, own};
1297 /// use ruststream_sqlx::{Extend, InboxSpec, InboxTable};
1298 /// use sqlx::{PgConnection, Postgres};
1299 /// # use ruststream_sqlx::PayloadRow;
1300 /// # use ruststream_sqlx::prelude::*;
1301 /// # use serde::Deserialize;
1302 /// # use sqlx::PgPool;
1303 ///
1304 /// #[derive(sqlx::FromRow)]
1305 /// pub struct Transcode {
1306 /// id: i64,
1307 /// payload: Vec<u8>,
1308 /// }
1309 ///
1310 /// impl InboxTable for Transcode {
1311 /// type Id = i64;
1312 /// type Table = InboxSpec<(Lease<DateTime<Utc>>, Payload, own::Extend)>;
1313 /// const TABLE: Self::Table = InboxSpec::new("transcodes", Column::new("id").generated())
1314 /// .lease(Column::new("locked_until"))
1315 /// .payload(Column::new("payload"))
1316 /// .own::<own::Extend>();
1317 ///
1318 /// fn id(&self) -> &i64 {
1319 /// &self.id
1320 /// }
1321 /// }
1322 ///
1323 /// // Each extension also counts in `extensions`, which shows a transcode that runs too long.
1324 /// impl Extend<Postgres> for Transcode {
1325 /// async fn extend(
1326 /// conn: &mut PgConnection,
1327 /// id: &i64,
1328 /// held: &DateTime<Utc>,
1329 /// until: &DateTime<Utc>,
1330 /// ) -> Result<bool, sqlx::Error> {
1331 /// let extended = sqlx::query(
1332 /// "UPDATE transcodes SET locked_until = $1, extensions = extensions + 1 \
1333 /// WHERE id = $2 AND locked_until = $3",
1334 /// )
1335 /// .bind(until)
1336 /// .bind(id)
1337 /// .bind(held)
1338 /// .execute(conn)
1339 /// .await?;
1340 /// Ok(extended.rows_affected() == 1)
1341 /// }
1342 /// }
1343 /// # impl PayloadRow for Transcode {
1344 /// # type Column = Vec<u8>;
1345 /// # fn payload(&self) -> &[u8] { &self.payload }
1346 /// # }
1347 /// # #[derive(Deserialize)]
1348 /// # struct Video { file: String }
1349 /// # #[subscriber(InboxQueue::<Transcode>::new("transcodes"))]
1350 /// # async fn transcode(video: &Video) -> HandlerOutcome {
1351 /// # tracing::info!(file = %video.file, "transcoding");
1352 /// # HandlerOutcome::ack()
1353 /// # }
1354 /// # pub fn app(pool: PgPool) -> RustStream {
1355 /// # RustStream::new(AppInfo::new("video", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
1356 /// # b.include(transcode);
1357 /// # })
1358 /// # }
1359 /// # }
1360 /// # fn main() {}
1361 /// ```
1362 Extend, OwnExtend
1363 );
1364 setting!(
1365 /// The service's own [`Lock`](crate::Lock): set by
1366 /// [`InboxSpec::own`](crate::InboxSpec::own).
1367 ///
1368 /// # Examples
1369 ///
1370 /// The service's own lock comes with its own unlock, [`own::Unlock`](Unlock):
1371 ///
1372 /// ```
1373 /// # #[cfg(feature = "postgres")]
1374 /// # mod demo {
1375 /// use ruststream_sqlx::dialect::{Column, KeyPart};
1376 /// use ruststream_sqlx::spec::{Advisory, Payload, own};
1377 /// use ruststream_sqlx::{InboxSpec, InboxTable, Lock};
1378 /// use sqlx::{PgConnection, Postgres};
1379 /// # use ruststream_sqlx::{PayloadRow, Unlock};
1380 /// # use ruststream_sqlx::prelude::*;
1381 /// # use serde::Deserialize;
1382 /// # use sqlx::PgPool;
1383 ///
1384 /// #[derive(sqlx::FromRow)]
1385 /// pub struct Booking {
1386 /// id: i64,
1387 /// payload: Vec<u8>,
1388 /// }
1389 ///
1390 /// impl InboxTable for Booking {
1391 /// type Id = i64;
1392 /// type Table = InboxSpec<(Advisory, Payload, own::Lock, own::Unlock)>;
1393 /// const TABLE: Self::Table = InboxSpec::new("bookings", Column::new("id").generated())
1394 /// .advisory(&[KeyPart::Literal("room-"), KeyPart::Column("room")])
1395 /// .payload(Column::new("payload"))
1396 /// .own::<own::Lock>()
1397 /// .own::<own::Unlock>();
1398 ///
1399 /// fn id(&self) -> &i64 {
1400 /// &self.id
1401 /// }
1402 /// }
1403 ///
1404 /// // The front desk application locks a room as `(42, hashtext(key))`, so the queue takes that lock.
1405 /// impl Lock<Postgres> for Booking {
1406 /// async fn lock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
1407 /// sqlx::query_scalar("SELECT pg_try_advisory_lock(42, hashtext($1))")
1408 /// .bind(key)
1409 /// .fetch_one(conn)
1410 /// .await
1411 /// }
1412 /// }
1413 /// # impl Unlock<Postgres> for Booking {
1414 /// # async fn unlock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
1415 /// # sqlx::query_scalar("SELECT pg_advisory_unlock(42, hashtext($1))")
1416 /// # .bind(key)
1417 /// # .fetch_one(conn)
1418 /// # .await
1419 /// # }
1420 /// # }
1421 /// # impl PayloadRow for Booking {
1422 /// # type Column = Vec<u8>;
1423 /// # fn payload(&self) -> &[u8] { &self.payload }
1424 /// # }
1425 /// # #[derive(Deserialize)]
1426 /// # struct Request { guest: String }
1427 /// # #[subscriber(InboxQueue::<Booking>::new("bookings"))]
1428 /// # async fn book(request: &Request) -> HandlerOutcome {
1429 /// # tracing::info!(guest = %request.guest, "booking");
1430 /// # HandlerOutcome::ack()
1431 /// # }
1432 /// # pub fn app(pool: PgPool) -> RustStream {
1433 /// # RustStream::new(AppInfo::new("hotel", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
1434 /// # b.include(book);
1435 /// # })
1436 /// # }
1437 /// # }
1438 /// # fn main() {}
1439 /// ```
1440 Lock, OwnLock
1441 );
1442 setting!(
1443 /// The service's own [`Unlock`](crate::Unlock): set by
1444 /// [`InboxSpec::own`](crate::InboxSpec::own).
1445 ///
1446 /// # Examples
1447 ///
1448 /// The service's own unlock releases what its own lock, [`own::Lock`](Lock), took:
1449 ///
1450 /// ```
1451 /// # #[cfg(feature = "postgres")]
1452 /// # mod demo {
1453 /// use ruststream_sqlx::dialect::{Column, KeyPart};
1454 /// use ruststream_sqlx::spec::{Advisory, Payload, own};
1455 /// use ruststream_sqlx::{InboxSpec, InboxTable, Unlock};
1456 /// use sqlx::{PgConnection, Postgres};
1457 /// # use ruststream_sqlx::{Lock, PayloadRow};
1458 /// # use ruststream_sqlx::prelude::*;
1459 /// # use serde::Deserialize;
1460 /// # use sqlx::PgPool;
1461 ///
1462 /// #[derive(sqlx::FromRow)]
1463 /// pub struct Rebalance {
1464 /// id: i64,
1465 /// payload: Vec<u8>,
1466 /// }
1467 ///
1468 /// impl InboxTable for Rebalance {
1469 /// type Id = i64;
1470 /// type Table = InboxSpec<(Advisory, Payload, own::Lock, own::Unlock)>;
1471 /// const TABLE: Self::Table = InboxSpec::new("rebalances", Column::new("id").generated())
1472 /// .advisory(&[KeyPart::Literal("portfolio-"), KeyPart::Column("portfolio")])
1473 /// .payload(Column::new("payload"))
1474 /// .own::<own::Lock>()
1475 /// .own::<own::Unlock>();
1476 ///
1477 /// fn id(&self) -> &i64 {
1478 /// &self.id
1479 /// }
1480 /// }
1481 ///
1482 /// // The trading desk locks a portfolio as `(7, hashtext(key))`, and the unlock releases that lock.
1483 /// impl Unlock<Postgres> for Rebalance {
1484 /// async fn unlock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
1485 /// sqlx::query_scalar("SELECT pg_advisory_unlock(7, hashtext($1))")
1486 /// .bind(key)
1487 /// .fetch_one(conn)
1488 /// .await
1489 /// }
1490 /// }
1491 /// # impl Lock<Postgres> for Rebalance {
1492 /// # async fn lock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
1493 /// # sqlx::query_scalar("SELECT pg_try_advisory_lock(7, hashtext($1))")
1494 /// # .bind(key)
1495 /// # .fetch_one(conn)
1496 /// # .await
1497 /// # }
1498 /// # }
1499 /// # impl PayloadRow for Rebalance {
1500 /// # type Column = Vec<u8>;
1501 /// # fn payload(&self) -> &[u8] { &self.payload }
1502 /// # }
1503 /// # #[derive(Deserialize)]
1504 /// # struct Target { equity: f64 }
1505 /// # #[subscriber(InboxQueue::<Rebalance>::new("rebalances"))]
1506 /// # async fn rebalance(target: &Target) -> HandlerOutcome {
1507 /// # tracing::info!(equity = target.equity, "rebalancing");
1508 /// # HandlerOutcome::ack()
1509 /// # }
1510 /// # pub fn app(pool: PgPool) -> RustStream {
1511 /// # RustStream::new(AppInfo::new("portfolios", "1.0.0")).with_broker(SqlxBroker::new(pool), |b| {
1512 /// # b.include(rebalance);
1513 /// # })
1514 /// # }
1515 /// # }
1516 /// # fn main() {}
1517 /// ```
1518 Unlock, OwnUnlock
1519 );
1520}
1521
1522/// An event a service writes itself: a marker of [`own`], which
1523/// [`InboxSpec::own`](crate::InboxSpec::own) takes.
1524#[diagnostic::on_unimplemented(
1525 message = "`{Self}` is not an event of the service's own",
1526 label = "not an event",
1527 note = "name one of `spec::own::{{Claim, Fetch, Ack, Retry, RetryAfter, Discard, DeadLetter, \
1528 Extend, Lock, Unlock}}`"
1529)]
1530pub trait OwnEvent: Declaration {}
1531
1532macro_rules! own_events {
1533 ($($event:ident),*) => {
1534 $(impl OwnEvent for own::$event {})*
1535 };
1536}
1537
1538own_events!(
1539 Claim, Fetch, Ack, Retry, RetryAfter, Discard, DeadLetter, Extend, Lock, Unlock
1540);
1541
1542/// An isolation level or a SQLite mode as a [`level`] marker, which
1543/// [`InboxSpec::opens`](crate::InboxSpec::opens) takes.
1544#[diagnostic::on_unimplemented(
1545 message = "`{Self}` is not an isolation level or a SQLite mode",
1546 label = "not a level",
1547 note = "name one of the `dialect::level` markers"
1548)]
1549pub trait OpeningLevel {
1550 /// The opening the marker stands for.
1551 const OPENING: Opening;
1552}
1553
1554macro_rules! levels {
1555 ($($marker:ident => $opening:expr),*) => {
1556 $(impl OpeningLevel for level::$marker {
1557 const OPENING: Opening = $opening;
1558 })*
1559 };
1560}
1561
1562levels!(
1563 ReadUncommitted => Opening::Isolation(Isolation::ReadUncommitted),
1564 ReadCommitted => Opening::Isolation(Isolation::ReadCommitted),
1565 RepeatableRead => Opening::Isolation(Isolation::RepeatableRead),
1566 Serializable => Opening::Isolation(Isolation::Serializable),
1567 Deferred => Opening::Mode(Mode::Deferred),
1568 Immediate => Opening::Mode(Mode::Immediate),
1569 Exclusive => Opening::Mode(Mode::Exclusive)
1570);