ruststream_sqlx/inbox/broker.rs
1//! `SqlxBroker`: its configuration and the connection it makes; the connected form and the closed
2//! witness live in `connected`.
3
4use std::borrow::Cow;
5use std::fmt;
6use std::num::NonZeroU32;
7use std::sync::Arc;
8use std::time::Duration;
9
10use ruststream::Broker;
11use ruststream_sqlx_dialect::{Dialect, Opens};
12use sqlx::pool::PoolOptions;
13use sqlx::{Database, Pool};
14use tokio::runtime::Handle;
15
16use super::database::notify::{StartListening, start_listening};
17use super::database::{BuiltIn, BuiltInDialect, Notifies, QueueDatabase};
18use super::engine::Events;
19use super::error::SqlxBrokerError;
20use super::events::Publish;
21use super::publish::Routes;
22use super::{FormDialect, FormOn, PayloadRow};
23
24mod connected;
25#[cfg(any(
26 feature = "postgres",
27 feature = "mysql",
28 feature = "sqlite",
29 feature = "any"
30))]
31mod describe;
32mod source;
33
34pub(crate) use connected::Shared;
35pub use connected::{ClosedSqlxBroker, ConnectedSqlxBroker};
36pub(crate) use source::PoolOwner;
37use source::{ConnectOptionsOf, PoolSource, Target};
38
39/// How long a subscription waits between claims that found its queue empty, unless it names
40/// another interval.
41const DEFAULT_POLL_INTERVAL: Duration = Duration::from_secs(1);
42
43/// How long a claim leases a row of a table in the lease form, unless the subscription names
44/// another lease.
45const DEFAULT_LEASE: Duration = Duration::from_secs(30);
46
47/// The dialect a broker will build statements with: one built into the crate, picked from the
48/// connection `connect` checks, or the service's own.
49pub(crate) enum DialectChoice<DB: Database, D> {
50 Picked(fn(&DB::Connection) -> Result<D, SqlxBrokerError>),
51 Given(Arc<D>),
52}
53
54impl<DB: Database, D: Dialect> DialectChoice<DB, D> {
55 /// The dialect of the database `conn` reaches.
56 pub(crate) fn resolve(&self, conn: &DB::Connection) -> Result<Arc<D>, SqlxBrokerError> {
57 match self {
58 Self::Picked(pick) => pick(conn).map(Arc::new),
59 Self::Given(dialect) => Ok(Arc::clone(dialect)),
60 }
61 }
62
63 /// The dialect's name, once there is one to name.
64 fn name(&self) -> &'static str {
65 match self {
66 Self::Picked(_) => "built in, picked at connect",
67 Self::Given(dialect) => dialect.name(),
68 }
69 }
70}
71
72impl<DB: Database, D> Clone for DialectChoice<DB, D> {
73 fn clone(&self) -> Self {
74 match self {
75 Self::Picked(pick) => Self::Picked(*pick),
76 Self::Given(dialect) => Self::Given(Arc::clone(dialect)),
77 }
78 }
79}
80
81/// The built-in dialect of the database `conn` reaches, or the error that names a database no
82/// built-in dialect serves.
83fn built_in<DB: BuiltInDialect>(conn: &DB::Connection) -> Result<BuiltIn<DB>, SqlxBrokerError> {
84 DB::dialect(conn).ok_or_else(|| SqlxBrokerError::Backend {
85 backend: DB::backend(conn).to_owned(),
86 })
87}
88
89/// How a route reaches its table's form on the dialect the broker connects with.
90pub(crate) type FormOf<D> = fn(&Arc<D>) -> FormDialect;
91
92/// The inbox broker: task queues in the service's own tables, served through a sqlx pool.
93///
94/// [`from_url`](Self::from_url) and [`from_connect_options`](Self::from_connect_options) record
95/// the pool's settings and do no I/O, so the service builds the broker in its app's builder;
96/// [`connect`](Broker::connect) builds the pool, and
97/// [`shutdown`](ruststream::ConnectedBroker::shutdown) closes it. [`new`](Self::new) takes a pool
98/// the service shares with code of its own, and the broker never closes that one. `connect` takes
99/// one connection to check the database, and picks the built-in dialect by it where the service
100/// passed none. A message in
101/// work holds one of the pool's connections until it settles (in the lease form outside
102/// transactional mode only while it settles), and a publish takes another, so the pool is sized
103/// for both.
104/// Subscriptions read tables through [`InboxQueue`](crate::InboxQueue) descriptors; publishing
105/// writes them through [`Repository`](crate::Repository) policies or through the routes this
106/// builder records.
107///
108/// # Examples
109///
110/// ```no_run
111/// # #[cfg(all(feature = "postgres", feature = "chrono"))]
112/// # mod demo {
113/// use ruststream::OutgoingMessage;
114/// use ruststream::prelude::*;
115/// use ruststream_sqlx::{Inbox, InboxQueue, Publish, SqlxBroker};
116/// use serde::Deserialize;
117/// use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
118/// use sqlx::{PgConnection, Postgres};
119///
120/// #[derive(Inbox, sqlx::FromRow)]
121/// #[inbox(table = "email_jobs")]
122/// pub struct SendEmail {
123/// #[field(id, generated)]
124/// job_id: i64,
125/// #[field(group)]
126/// name: String,
127/// #[field(payload)]
128/// payload: Vec<u8>,
129/// }
130///
131/// impl Publish<Postgres> for SendEmail {
132/// async fn publish(
133/// conn: &mut PgConnection,
134/// message: &OutgoingMessage<'_>,
135/// ) -> Result<(), sqlx::Error> {
136/// sqlx::query("INSERT INTO email_jobs (name, payload) VALUES ($1, $2)")
137/// .bind(message.name())
138/// .bind(message.payload())
139/// .execute(conn)
140/// .await?;
141/// Ok(())
142/// }
143/// }
144///
145/// #[derive(Deserialize)]
146/// struct Email {
147/// to: String,
148/// }
149///
150/// #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
151/// async fn send(email: &Email) -> HandlerOutcome {
152/// tracing::info!(to = %email.to, "sending");
153/// HandlerOutcome::ack()
154/// }
155///
156/// #[ruststream::app]
157/// fn app() -> impl App {
158/// // The options read the `PG*` environment variables; the broker builds the pool when the
159/// // service starts and closes it when the service stops.
160/// let broker = SqlxBroker::from_connect_options(PgConnectOptions::new(), PgPoolOptions::new());
161/// RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(
162/// broker.route::<SendEmail>("emails"),
163/// |b| {
164/// b.include(send);
165/// },
166/// )
167/// }
168/// # }
169/// # fn main() {}
170/// ```
171///
172/// # Testing
173///
174/// A test hands [`TestApp::start`] the app `main` runs, unchanged, and addresses the broker by its
175/// production type (feature `testing`). The broker connects in process to the database the test's
176/// pool reaches, so the service's SQL runs as in production, and the harness waits for every
177/// publish: a row counts for the queue it lands in, its table and group, whatever name a
178/// subscription reads it by. SQLite in memory needs no server. After `tokio::time::pause()` the queue reads "now"
179/// from the test's clock, which `tb.advance(d)` moves past a delayed retry or a lease; a table on
180/// [`DatabaseClock`](crate::DatabaseClock) reads the database's time instead. The crate registers
181/// `SqlxBroker<Postgres>`, `SqlxBroker<MySql>`, `SqlxBroker<Sqlite>` and `SqlxBroker<Any>` with the
182/// harness, and a broker on a dialect of the service's own is registered by the service
183/// ([`with_dialect`](Self::with_dialect)).
184///
185/// [`TestApp::start`]: https://docs.rs/ruststream/latest/ruststream/testing/struct.TestApp.html#method.start
186///
187/// ```no_run
188/// # #[cfg(all(feature = "sqlite", feature = "chrono", feature = "testing"))]
189/// # mod demo {
190/// use std::error::Error;
191///
192/// use chrono::{DateTime, Utc};
193/// use ruststream::OutgoingMessage;
194/// use ruststream::testing::TestApp;
195/// use ruststream_sqlx::prelude::*;
196/// use serde::{Deserialize, Serialize};
197/// use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
198/// use sqlx::{Connection, Sqlite, SqliteConnection, SqlitePool};
199///
200/// #[derive(Inbox, sqlx::FromRow)]
201/// #[inbox(table = "thumbnail_jobs")]
202/// pub struct MakeThumbnail {
203/// #[field(id, generated)]
204/// id: i64,
205/// #[field(attempt, generated)]
206/// attempt: i16,
207/// #[field(locked_until)]
208/// locked_until: Option<DateTime<Utc>>,
209/// #[field(payload)]
210/// payload: Vec<u8>,
211/// }
212///
213/// impl Publish<Sqlite> for MakeThumbnail {
214/// async fn publish(
215/// conn: &mut SqliteConnection,
216/// message: &OutgoingMessage<'_>,
217/// ) -> Result<(), sqlx::Error> {
218/// sqlx::query("INSERT INTO thumbnail_jobs (payload) VALUES (?)")
219/// .bind(message.payload())
220/// .execute(conn)
221/// .await?;
222/// Ok(())
223/// }
224/// }
225///
226/// #[derive(Serialize, Deserialize, Outgoing)]
227/// pub struct Image {
228/// path: String,
229/// }
230///
231/// #[subscriber(InboxQueue::<MakeThumbnail>::new("thumbnails"))]
232/// async fn thumbnail(image: &Image) -> HandlerOutcome {
233/// tracing::info!(path = %image.path, "making a thumbnail");
234/// HandlerOutcome::ack()
235/// }
236///
237/// pub fn app(pool: SqlitePool) -> RustStream {
238/// RustStream::new(AppInfo::new("thumbnails", "0.1.0")).with_broker(
239/// SqlxBroker::new(pool).route::<MakeThumbnail>("thumbnails"),
240/// |b| {
241/// b.include(thumbnail);
242/// },
243/// )
244/// }
245///
246/// const SCHEMA: &str = "CREATE TABLE thumbnail_jobs (id INTEGER PRIMARY KEY, \
247/// attempt INTEGER NOT NULL DEFAULT 1, locked_until TEXT, payload BLOB NOT NULL)";
248///
249/// // The body of a `#[tokio::test]`.
250/// pub async fn a_thumbnail_is_made() -> Result<(), Box<dyn Error>> {
251/// // A database in memory that every connection naming it shares, alive while one of them is
252/// // open: the test holds that one beside the pool.
253/// let options: SqliteConnectOptions =
254/// "sqlite:file:thumbnails?mode=memory&cache=shared".parse()?;
255/// let _keeper = SqliteConnection::connect_with(&options).await?;
256/// let pool = SqlitePoolOptions::new()
257/// .min_connections(1)
258/// .idle_timeout(None)
259/// .connect_with(options)
260/// .await?;
261/// sqlx::raw_sql(SCHEMA).execute(&pool).await?;
262///
263/// let tb = TestApp::start(app(pool)).await?;
264/// // Through the route and the service's own `Publish`, as in production.
265/// tb.broker::<SqlxBroker<Sqlite>>()
266/// .message(&Image { path: "cat.png".to_owned() })
267/// .to("thumbnails")
268/// .publish()
269/// .await?;
270/// tb.broker::<SqlxBroker<Sqlite>>()
271/// .subscriber("thumbnails")
272/// .assert_called_once()
273/// .settled(HandlerOutcome::ack());
274/// Ok(())
275/// }
276/// # }
277/// # fn main() {}
278/// ```
279pub struct SqlxBroker<DB: Database, D = BuiltIn<DB>> {
280 pub(crate) pool: PoolSource<DB>,
281 pub(crate) dialect: DialectChoice<DB, D>,
282 routes: Routes<DB, FormOf<D>>,
283 poll_interval: Duration,
284 lease: Duration,
285 connection_limit: Option<NonZeroU32>,
286 /// The start of the listening connection, with `listen_notify`.
287 pub(crate) listen: Option<StartListening<DB>>,
288}
289
290impl<DB: Database, D: Dialect> fmt::Debug for SqlxBroker<DB, D> {
291 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
292 f.debug_struct("SqlxBroker")
293 .field("pool", &self.pool.kind())
294 .field("dialect", &self.dialect.name())
295 .field("routes", &self.routes)
296 .field("poll_interval", &self.poll_interval)
297 .field("lease", &self.lease)
298 .field("connection_limit", &self.connection_limit)
299 .field("listen_notify", &self.listen.is_some())
300 .finish_non_exhaustive()
301 }
302}
303
304impl<DB: BuiltInDialect> SqlxBroker<DB, BuiltIn<DB>> {
305 /// A broker that builds its pool from `url` and `pool` when it connects, with the dialect
306 /// built into the crate for the database.
307 ///
308 /// Synchronous and free of I/O, so the service builds it in its app's builder, before the
309 /// runtime starts. [`connect`](Broker::connect) builds the pool, takes one connection to
310 /// check the database, and picks the dialect by it;
311 /// [`shutdown`](ruststream::ConnectedBroker::shutdown) closes the pool once the broker's own
312 /// work is done. The driver reads the URL at `connect`: a URL it cannot read stops the
313 /// service's start with [`SqlxBrokerError::Url`]. A pool the service also queries outside
314 /// its handlers is the service's own: [`new`](Self::new) takes it.
315 ///
316 /// # Examples
317 ///
318 /// ```
319 /// # #[cfg(feature = "postgres")]
320 /// # mod demo {
321 /// use std::env;
322 ///
323 /// use ruststream_sqlx::prelude::*;
324 /// use serde::Deserialize;
325 /// use sqlx::postgres::PgPoolOptions;
326 ///
327 /// #[derive(Inbox, sqlx::FromRow)]
328 /// #[inbox(table = "email_jobs")]
329 /// pub struct SendEmail {
330 /// #[field(id, generated)]
331 /// job_id: i64,
332 /// #[field(group)]
333 /// name: String,
334 /// #[field(payload)]
335 /// payload: Vec<u8>,
336 /// }
337 ///
338 /// #[derive(Deserialize)]
339 /// struct Email {
340 /// to: String,
341 /// }
342 ///
343 /// #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
344 /// async fn send(email: &Email) -> HandlerOutcome {
345 /// tracing::info!(to = %email.to, "sending");
346 /// HandlerOutcome::ack()
347 /// }
348 ///
349 /// #[ruststream::app]
350 /// fn app() -> impl App {
351 /// let url = env::var("DATABASE_URL").unwrap_or_else(|_| "postgres://localhost/mail".into());
352 /// // Nothing connects here: the broker builds the pool when the service starts.
353 /// let broker = SqlxBroker::from_url(url, PgPoolOptions::new().max_connections(8));
354 /// RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
355 /// b.include(send);
356 /// })
357 /// }
358 /// # }
359 /// # fn main() {}
360 /// ```
361 #[must_use]
362 pub fn from_url(url: impl Into<String>, pool: PoolOptions<DB>) -> Self {
363 let target = Target::Url(url.into());
364 Self::built(
365 PoolSource::Built { target, pool },
366 DialectChoice::Picked(built_in::<DB>),
367 )
368 }
369
370 /// A broker that builds its pool from the driver's connect options and `pool` when it
371 /// connects, with the dialect built into the crate for the database.
372 ///
373 /// The options a service sets in code or reads from the environment
374 /// (`PgConnectOptions::new()` reads the `PG*` variables); otherwise as
375 /// [`from_url`](Self::from_url): synchronous, free of I/O, the pool built at `connect` and
376 /// closed at `shutdown`.
377 ///
378 /// # Examples
379 ///
380 /// ```
381 /// # #[cfg(feature = "postgres")]
382 /// # mod demo {
383 /// use ruststream_sqlx::prelude::*;
384 /// use serde::Deserialize;
385 /// use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
386 ///
387 /// #[derive(Inbox, sqlx::FromRow)]
388 /// #[inbox(table = "email_jobs")]
389 /// pub struct SendEmail {
390 /// #[field(id, generated)]
391 /// job_id: i64,
392 /// #[field(group)]
393 /// name: String,
394 /// #[field(payload)]
395 /// payload: Vec<u8>,
396 /// }
397 ///
398 /// #[derive(Deserialize)]
399 /// struct Email {
400 /// to: String,
401 /// }
402 ///
403 /// #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
404 /// async fn send(email: &Email) -> HandlerOutcome {
405 /// tracing::info!(to = %email.to, "sending");
406 /// HandlerOutcome::ack()
407 /// }
408 ///
409 /// #[ruststream::app]
410 /// fn app() -> impl App {
411 /// // The options read the `PG*` environment variables; the application name is the
412 /// // service's own.
413 /// let options = PgConnectOptions::new().application_name("mailer");
414 /// let broker = SqlxBroker::from_connect_options(options, PgPoolOptions::new());
415 /// RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
416 /// b.include(send);
417 /// })
418 /// }
419 /// # }
420 /// # fn main() {}
421 /// ```
422 #[must_use]
423 pub fn from_connect_options(options: ConnectOptionsOf<DB>, pool: PoolOptions<DB>) -> Self {
424 let target = Target::Options(options);
425 Self::built(
426 PoolSource::Built { target, pool },
427 DialectChoice::Picked(built_in::<DB>),
428 )
429 }
430
431 /// A broker on the service's own `pool`, with the dialect built into the crate for its
432 /// database.
433 ///
434 /// For a pool the service shares with the rest of its code: an HTTP endpoint that writes
435 /// rows, a migration run before the app. The broker never closes it; the service does, once
436 /// the app has stopped. Synchronous and free of I/O: the database type comes from the pool,
437 /// and nothing connects before [`connect`](Broker::connect), which picks the dialect from the
438 /// connection it checks. An `AnyPool` reaches a database named only then, and a database
439 /// whose dialect's feature is off stops `connect` with [`SqlxBrokerError::Backend`]. sqlx
440 /// builds a pool only inside a Tokio runtime, so a service that owns one builds it in its
441 /// own `main`; a broker that builds its pool itself is [`from_url`](Self::from_url).
442 ///
443 /// # Examples
444 ///
445 /// ```
446 /// # #[cfg(feature = "postgres")]
447 /// # mod demo {
448 /// use std::error::Error;
449 ///
450 /// use ruststream_sqlx::prelude::*;
451 /// use serde::Deserialize;
452 /// use sqlx::postgres::{PgConnectOptions, PgPool, PgPoolOptions};
453 ///
454 /// #[derive(Inbox, sqlx::FromRow)]
455 /// #[inbox(table = "email_jobs")]
456 /// pub struct SendEmail {
457 /// #[field(id, generated)]
458 /// job_id: i64,
459 /// #[field(group)]
460 /// name: String,
461 /// #[field(payload)]
462 /// payload: Vec<u8>,
463 /// }
464 ///
465 /// #[derive(Deserialize)]
466 /// struct Email {
467 /// to: String,
468 /// }
469 ///
470 /// #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
471 /// async fn send(email: &Email) -> HandlerOutcome {
472 /// tracing::info!(to = %email.to, "sending");
473 /// HandlerOutcome::ack()
474 /// }
475 ///
476 /// pub fn app(pool: PgPool) -> RustStream {
477 /// RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
478 /// b.include(send);
479 /// })
480 /// }
481 ///
482 /// // The service shares the pool with code of its own, so it builds the pool in its `main`,
483 /// // inside the runtime, and closes it after the app.
484 /// #[tokio::main]
485 /// pub async fn main() -> Result<(), Box<dyn Error>> {
486 /// // A lazy pool: nothing connects before the service starts. The options read the
487 /// // `PG*` environment variables.
488 /// let pool = PgPoolOptions::new().connect_lazy_with(PgConnectOptions::new());
489 /// let app = app(pool.clone());
490 /// # // The example runs without a database: it stops before the app connects.
491 /// # drop(app);
492 /// # return Ok(());
493 /// app.run().await?;
494 /// pool.close().await;
495 /// Ok(())
496 /// }
497 /// # }
498 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
499 /// # #[cfg(feature = "postgres")]
500 /// # demo::main()?;
501 /// # Ok(())
502 /// # }
503 /// ```
504 #[must_use]
505 pub fn new(pool: Pool<DB>) -> Self {
506 Self::built(
507 PoolSource::Service(pool),
508 DialectChoice::Picked(built_in::<DB>),
509 )
510 }
511}
512
513impl<DB: QueueDatabase, D: Dialect + 'static> SqlxBroker<DB, D> {
514 /// A broker on `pool` whose statements `dialect`, a dialect of the service's own, builds: a
515 /// driver this crate has no dialect for, or a built-in dialect the service wraps to write a
516 /// statement its own way.
517 ///
518 /// The dialect's type is the broker's second type parameter, so what a subscription asks of
519 /// it is checked where the subscription mounts: a table in the row lock form needs the
520 /// dialect to implement [`RowLock`](crate::dialect::RowLock), a lease table
521 /// [`Lease`](crate::dialect::Lease), and a table that names an isolation level or a SQLite
522 /// mode [`Opens`] for it. The dialect runs while subscriptions start; messages travel through
523 /// the statements it built.
524 ///
525 /// The crate registers the brokers of its built-in dialects with the test harness. A test of
526 /// a broker on a dialect of the service's own registers its type once,
527 /// `ruststream::register_testable_broker!(SqlxBroker<Postgres, Audited>);`, and addresses it
528 /// by that type, `tb.broker::<SqlxBroker<Postgres, Audited>>()`.
529 ///
530 /// # Examples
531 ///
532 /// ```
533 /// # #[cfg(feature = "postgres")]
534 /// # mod demo {
535 /// # use std::num::NonZeroUsize;
536 /// # use ruststream_sqlx::dialect::{self, ClaimShape, Dialect, Opening, RowLock, Statement, StatementError, TableName, TableSpec};
537 /// use ruststream_sqlx::prelude::*;
538 /// use serde::Deserialize;
539 /// use sqlx::PgPool;
540 ///
541 /// /// Postgres with statements of the service's own.
542 /// #[derive(Debug)]
543 /// pub struct Audited;
544 /// # impl Dialect for Audited {
545 /// # fn name(&self) -> &'static str { "audited" }
546 /// # fn quote_into(&self, ident: &str, out: &mut String) { dialect::Postgres.quote_into(ident, out) }
547 /// # fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) { dialect::Postgres.placeholder_into(index, out) }
548 /// # fn fetch(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.fetch(spec) }
549 /// # fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.ack(spec) }
550 /// # fn retry(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { dialect::Postgres.retry(spec) }
551 /// # fn retry_after(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.retry_after(spec) }
552 /// # fn discard(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.discard(spec) }
553 /// # fn dead_letter_group(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.dead_letter_group(spec) }
554 /// # fn dead_letter_table(&self, spec: &TableSpec<'_>, target: TableName<'_>) -> Result<Vec<Statement>, StatementError> { dialect::Postgres.dead_letter_table(spec, target) }
555 /// # fn insert(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> { dialect::Postgres.insert(spec) }
556 /// # fn begin(&self, opening: Opening) -> Result<Option<&'static str>, StatementError> { dialect::Postgres.begin(opening) }
557 /// # fn fifo_guard(&self, spec: &TableSpec<'_>) -> Result<Option<Statement>, StatementError> { dialect::Postgres.fifo_guard(spec) }
558 /// # }
559 /// # impl RowLock for Audited {
560 /// # fn lock_claim(&self, spec: &TableSpec<'_>, shape: ClaimShape) -> Result<Statement, StatementError> { dialect::Postgres.lock_claim(spec, shape) }
561 /// # }
562 ///
563 /// #[derive(Inbox, sqlx::FromRow)]
564 /// #[inbox(table = "email_jobs")]
565 /// pub struct SendEmail {
566 /// #[field(id, generated)]
567 /// job_id: i64,
568 /// #[field(group)]
569 /// name: String,
570 /// #[field(payload)]
571 /// payload: Vec<u8>,
572 /// }
573 ///
574 /// #[derive(Deserialize)]
575 /// struct Email {
576 /// to: String,
577 /// }
578 ///
579 /// #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
580 /// async fn send(email: &Email) -> HandlerOutcome {
581 /// tracing::info!(to = %email.to, "sending");
582 /// HandlerOutcome::ack()
583 /// }
584 ///
585 /// pub fn app(pool: PgPool) -> RustStream {
586 /// // A `SqlxBroker<Postgres, Audited>`: `send` mounts only because `Audited` implements
587 /// // `RowLock`, the form of `email_jobs`.
588 /// RustStream::new(AppInfo::new("mailer", "0.1.0"))
589 /// .with_broker(SqlxBroker::with_dialect(pool, Audited), |b| {
590 /// b.include(send);
591 /// })
592 /// }
593 /// # }
594 /// # fn main() {}
595 /// ```
596 #[must_use]
597 pub fn with_dialect(pool: Pool<DB>, dialect: D) -> Self {
598 Self::built(
599 PoolSource::Service(pool),
600 DialectChoice::Given(Arc::new(dialect)),
601 )
602 }
603
604 fn built(pool: PoolSource<DB>, dialect: DialectChoice<DB, D>) -> Self {
605 Self {
606 pool,
607 dialect,
608 routes: Routes::default(),
609 poll_interval: DEFAULT_POLL_INTERVAL,
610 lease: DEFAULT_LEASE,
611 connection_limit: None,
612 listen: None,
613 }
614 }
615
616 /// Leads publishes to `name` into the table of `Row`, through its [`Publish`], and opens a
617 /// subscription by that name (`#[subscriber("name")]`) on the same table.
618 ///
619 /// A name ending in `*` leads every name that starts with what precedes it; an exact name
620 /// wins over a prefix, and a longer prefix over a shorter one. Routing a name twice leads it
621 /// to the last row type. The broker's default publisher is this route table: a reply or an
622 /// `Out` slot without a policy of its own publishes through it.
623 ///
624 /// A route asks of the broker's dialect what an [`InboxQueue`](crate::InboxQueue) of the row
625 /// asks: the table's form, and the isolation level or SQLite mode the struct declares. A
626 /// by-name subscription names no row type, so the route is where its table is checked.
627 ///
628 /// # Examples
629 ///
630 /// ```
631 /// # #[cfg(feature = "postgres")]
632 /// # mod demo {
633 /// use ruststream::OutgoingMessage;
634 /// use ruststream_sqlx::prelude::*;
635 /// use serde::{Deserialize, Serialize};
636 /// use sqlx::{PgConnection, PgPool, Postgres};
637 ///
638 /// #[derive(Inbox, sqlx::FromRow)]
639 /// #[inbox(table = "report_jobs")]
640 /// pub struct Report {
641 /// #[field(id, generated)]
642 /// id: i64,
643 /// #[field(group)]
644 /// name: String,
645 /// #[field(payload)]
646 /// payload: Vec<u8>,
647 /// }
648 ///
649 /// impl Publish<Postgres> for Report {
650 /// async fn publish(
651 /// conn: &mut PgConnection,
652 /// message: &OutgoingMessage<'_>,
653 /// ) -> Result<(), sqlx::Error> {
654 /// sqlx::query("INSERT INTO report_jobs (name, payload) VALUES ($1, $2)")
655 /// .bind(message.name())
656 /// .bind(message.payload())
657 /// .execute(conn)
658 /// .await?;
659 /// Ok(())
660 /// }
661 /// }
662 ///
663 /// #[derive(Deserialize)]
664 /// struct Daily {
665 /// day: u32,
666 /// }
667 ///
668 /// #[derive(Serialize, Outgoing)]
669 /// #[outgoing(name = "reports.weekly")]
670 /// struct Weekly {
671 /// week: u32,
672 /// }
673 ///
674 /// // Read by name from `report_jobs`, answered with a row of `report_jobs` in group `reports.weekly`.
675 /// #[subscriber("reports.daily", reply)]
676 /// async fn roll_up(daily: &Daily) -> Weekly {
677 /// Weekly { week: daily.day / 7 }
678 /// }
679 ///
680 /// pub fn app(pool: PgPool) -> RustStream {
681 /// // `reports.daily` and `reports.weekly` both lead into `report_jobs`.
682 /// let broker = SqlxBroker::new(pool).route::<Report>("reports.*");
683 /// RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(broker, |b| {
684 /// b.include(roll_up);
685 /// })
686 /// }
687 /// # }
688 /// # fn main() {}
689 /// ```
690 #[must_use]
691 pub fn route<Row>(mut self, name: impl Into<Cow<'static, str>>) -> Self
692 where
693 Row: Publish<DB> + Events<DB> + PayloadRow,
694 Row::Form: FormOn<D>,
695 D: Opens<Row::Opening>,
696 {
697 self.routes
698 .add::<Row>(name.into(), <Row::Form as FormOn<D>>::erase);
699 self
700 }
701
702 /// How long a subscription waits between claims that found its queue empty; one second
703 /// unless set. A subscription's own [`poll_interval`](crate::InboxQueue::poll_interval)
704 /// overrides it. A publish through this broker wakes the subscriptions of its table and group
705 /// before then, and so does a notification with [`listen_notify`](Self::listen_notify)
706 /// ([waking a subscription](https://powersemmi.github.io/ruststream-sqlx/latest/components/wake/)).
707 ///
708 /// # Examples
709 ///
710 /// ```
711 /// # #[cfg(feature = "postgres")]
712 /// # mod demo {
713 /// use std::time::Duration;
714 ///
715 /// use ruststream_sqlx::prelude::*;
716 /// use serde::Deserialize;
717 /// use sqlx::PgPool;
718 ///
719 /// #[derive(Inbox, sqlx::FromRow)]
720 /// #[inbox(table = "email_jobs")]
721 /// pub struct SendEmail {
722 /// #[field(id, generated)]
723 /// job_id: i64,
724 /// #[field(group)]
725 /// name: String,
726 /// #[field(payload)]
727 /// payload: Vec<u8>,
728 /// }
729 ///
730 /// #[derive(Deserialize)]
731 /// struct Email {
732 /// to: String,
733 /// }
734 ///
735 /// #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
736 /// async fn send(email: &Email) -> HandlerOutcome {
737 /// tracing::info!(to = %email.to, "sending");
738 /// HandlerOutcome::ack()
739 /// }
740 ///
741 /// pub fn app(pool: PgPool) -> RustStream {
742 /// // Every queue on this broker checks twice a second when it runs dry.
743 /// let broker = SqlxBroker::new(pool).poll_interval(Duration::from_millis(500));
744 /// RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
745 /// b.include(send);
746 /// })
747 /// }
748 /// # }
749 /// # fn main() {}
750 /// ```
751 #[must_use]
752 pub const fn poll_interval(mut self, interval: Duration) -> Self {
753 self.poll_interval = interval;
754 self
755 }
756
757 /// How long a claim leases a row of a table in the lease form (one with a
758 /// `#[field(locked_until)]` field); thirty seconds unless set. A subscription's own
759 /// [`lease`](crate::InboxQueue::lease) overrides it, and a table in another form ignores it.
760 ///
761 /// A lease is whole seconds, at least one: a shorter one is rounded up. The subscription
762 /// extends the lease of every delivery in work each half lease, so a handler may run longer
763 /// than its lease; the lease is how long a row stays out of the queue after its process
764 /// crashed. A delivery dropped unsettled releases its row at once. A lease runs out under a
765 /// running handler only when its extensions fail or stop (the database out of reach, the
766 /// subscription closed, the broker [shut down](ruststream::ConnectedBroker::shutdown)): the
767 /// row then goes to the next claim, and the late settlement fails with
768 /// [`SqlxBrokerError::LeaseLost`].
769 ///
770 /// # Examples
771 ///
772 /// ```
773 /// # #[cfg(all(feature = "postgres", feature = "chrono"))]
774 /// # mod demo {
775 /// use std::time::Duration;
776 ///
777 /// use chrono::{DateTime, Utc};
778 /// use ruststream_sqlx::prelude::*;
779 /// use serde::Deserialize;
780 /// use sqlx::PgPool;
781 ///
782 /// #[derive(Inbox, sqlx::FromRow)]
783 /// #[inbox(table = "video_jobs")]
784 /// pub struct Transcode {
785 /// #[field(id, generated)]
786 /// id: i64,
787 /// #[field(locked_until)]
788 /// locked_until: Option<DateTime<Utc>>,
789 /// #[field(payload)]
790 /// payload: Vec<u8>,
791 /// }
792 ///
793 /// #[derive(Deserialize)]
794 /// struct Video {
795 /// path: String,
796 /// }
797 ///
798 /// #[subscriber(InboxQueue::<Transcode>::new("videos"))]
799 /// async fn transcode(video: &Video) -> HandlerOutcome {
800 /// tracing::info!(path = %video.path, "transcoding");
801 /// HandlerOutcome::ack()
802 /// }
803 ///
804 /// pub fn app(pool: PgPool) -> RustStream {
805 /// // A video whose process crashed goes back to the queue once its ten-second lease runs out.
806 /// let broker = SqlxBroker::new(pool).lease(Duration::from_secs(10));
807 /// RustStream::new(AppInfo::new("transcoder", "0.1.0")).with_broker(broker, |b| {
808 /// b.include(transcode);
809 /// })
810 /// }
811 /// # }
812 /// # fn main() {}
813 /// ```
814 #[must_use]
815 pub const fn lease(mut self, lease: Duration) -> Self {
816 self.lease = lease;
817 self
818 }
819
820 /// The most connections the service opens on the database: the pool's size, and the
821 /// connections of the dedicated threads of every subscription mounted with
822 /// [`on_threads`](crate::InboxSettings::on_threads). No limit unless set.
823 ///
824 /// Each such subscription counts its threads times their connections when it opens. One
825 /// that would take the sum past the limit refuses to open, with
826 /// [`SqlxBrokerError::ConnectionLimit`] naming the numbers, and the service does not start.
827 /// Set it to what the database lets this service open.
828 ///
829 /// # Examples
830 ///
831 /// ```
832 /// # #[cfg(feature = "postgres")]
833 /// # mod demo {
834 /// use ruststream::nonzero;
835 /// use ruststream_sqlx::prelude::*;
836 /// use sqlx::{PgPool, Postgres};
837 /// # #[derive(Inbox, sqlx::FromRow)]
838 /// # #[inbox(table = "scan_jobs")]
839 /// # pub struct ScanJob { #[field(id)] id: i64, #[field(payload)] payload: Vec<u8> }
840 /// # #[derive(serde::Deserialize)]
841 /// # struct Image { path: String }
842 ///
843 /// #[subscriber(InboxQueue::<ScanJob>::new("scans"))]
844 /// async fn scan(image: &Image, Ctx(pool): Ctx<keys::Pool<Postgres>>) -> HandlerOutcome {
845 /// let noted = sqlx::query("INSERT INTO scanned (path) VALUES ($1)")
846 /// .bind(&image.path)
847 /// .execute(&pool)
848 /// .await;
849 /// if noted.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
850 /// }
851 ///
852 /// pub fn app(pool: PgPool) -> RustStream {
853 /// // The database lets this service open 20 connections: the pool's and 4 threads' own.
854 /// let broker = SqlxBroker::new(pool).connection_limit(nonzero!(20u32));
855 /// RustStream::new(AppInfo::new("scanner", "0.1.0")).with_broker(broker, |b| {
856 /// b.include(scan.on_threads(InboxThreads::new(nonzero!(4))));
857 /// })
858 /// }
859 /// # }
860 /// # fn main() {}
861 /// ```
862 #[must_use]
863 pub const fn connection_limit(mut self, limit: NonZeroU32) -> Self {
864 self.connection_limit = Some(limit);
865 self
866 }
867
868 /// Wakes the broker's subscriptions on Postgres notifications: a row another process announced
869 /// with `pg_notify` is claimed at once, not on the next poll.
870 ///
871 /// The broker holds one connection of the pool for its life, from
872 /// [`connect`](Broker::connect) to [`shutdown`](ruststream::ConnectedBroker::shutdown), and
873 /// listens there.
874 /// Each subscription listens on the channel of its table, qualified with its schema, as it
875 /// opens; a notification whose payload names a group wakes that group's subscription, and one
876 /// with an empty payload wakes every subscription of the table. Each publish of the broker
877 /// sends `SELECT pg_notify(table, group)` on the connection that wrote the row, right after
878 /// the write: one more statement per publish. A publish whose notification fails is still a
879 /// publish; the row waits for the poll interval and the failure is logged.
880 ///
881 /// The poll interval stays: a notification sent while the listening connection was lost is
882 /// gone, so after a reconnect every subscription claims once, and a row written by the
883 /// service's own SQL or a handler's [`Tx`](crate::Tx) wakes nobody unless the service sends
884 /// the notification itself. It is opt-in because of its cost: the pool connection it holds,
885 /// the statement each publish adds, and the lock, global to the server, that Postgres takes
886 /// at the commit of each transaction that notifies, so those commits run one after another,
887 /// which limits many concurrent writers.
888 ///
889 /// A table whose qualified name is longer than Postgres's 63-byte channel names stops its
890 /// subscription at startup with [`SqlxBrokerError::Declaration`].
891 ///
892 /// # Examples
893 ///
894 /// ```
895 /// # #[cfg(feature = "postgres")]
896 /// # mod demo {
897 /// use std::time::Duration;
898 ///
899 /// use ruststream_sqlx::prelude::*;
900 /// use serde::Deserialize;
901 /// use sqlx::PgPool;
902 ///
903 /// #[derive(Inbox, sqlx::FromRow)]
904 /// #[inbox(table = "email_jobs")]
905 /// pub struct SendEmail {
906 /// #[field(id, generated)]
907 /// job_id: i64,
908 /// #[field(group)]
909 /// name: String,
910 /// #[field(payload)]
911 /// payload: Vec<u8>,
912 /// }
913 ///
914 /// #[derive(Deserialize)]
915 /// struct Email {
916 /// to: String,
917 /// }
918 ///
919 /// #[subscriber(InboxQueue::<SendEmail>::new("emails"))]
920 /// async fn send(email: &Email) -> HandlerOutcome {
921 /// tracing::info!(to = %email.to, "sending");
922 /// HandlerOutcome::ack()
923 /// }
924 ///
925 /// pub fn app(pool: PgPool) -> RustStream {
926 /// // A row another service writes and announces with
927 /// // `SELECT pg_notify('email_jobs', 'emails')` is sent at once; the poll each minute
928 /// // catches a row whose notification was lost.
929 /// let broker = SqlxBroker::new(pool)
930 /// .poll_interval(Duration::from_secs(60))
931 /// .listen_notify();
932 /// RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
933 /// b.include(send);
934 /// })
935 /// }
936 /// # }
937 /// # fn main() {}
938 /// ```
939 #[must_use]
940 pub fn listen_notify(mut self) -> Self
941 where
942 DB: Notifies,
943 {
944 self.listen = Some(start_listening::<DB>());
945 self
946 }
947}
948
949impl<DB: QueueDatabase, D: Dialect + 'static> Broker for SqlxBroker<DB, D> {
950 type Error = SqlxBrokerError;
951 type Connected = ConnectedSqlxBroker<DB, D>;
952
953 async fn connect(self) -> Result<Self::Connected, Self::Error> {
954 let (pool, owner) = self.pool.open()?;
955 let conn = pool
956 .acquire()
957 .await
958 .map_err(|source| SqlxBrokerError::Connect { source })?;
959 let dialect = self.dialect.resolve(&conn)?;
960 drop(conn);
961 let listening = match self.listen {
962 Some(start) => Some(
963 start(pool.clone())
964 .await
965 .map_err(|source| SqlxBrokerError::Connect { source })?,
966 ),
967 None => None,
968 };
969 Ok(ConnectedSqlxBroker::new(
970 self,
971 (pool, owner),
972 dialect,
973 Handle::current(),
974 listening,
975 ))
976 }
977}