ruststream_sqlx/inbox/
named.rs

1//! By-name subscriptions: `#[subscriber("emails")]` and [`Subscribe`], through the route a name
2//! takes.
3//!
4//! A route's row that leaves every event to the crate is read by its role columns into
5//! [`NamedRow`], one concrete type whatever the route; any other row is read by its own code and
6//! erased behind a box.
7
8mod by_name;
9mod database;
10pub(crate) mod kinds;
11mod row;
12
13use std::fmt;
14use std::pin::Pin;
15use std::sync::Arc;
16use std::time::Duration;
17
18use futures::future::{BoxFuture, Either};
19use futures::{Stream, StreamExt};
20use ruststream::codec::CodecError;
21use ruststream::{
22    AckError, BrokerMoves, HeaderMap, IncomingMessage, RetryDeclaration, Subscribe, Subscriber,
23};
24use sync_wrapper::SyncWrapper;
25
26pub use by_name::ByName;
27pub use database::RoleColumns;
28pub use row::{NamedBytes, NamedId, NamedRow, NamedTime};
29
30use super::broker::{ConnectedSqlxBroker, Shared};
31use super::database::QueueDatabase;
32use super::delivery::InboxDelivery;
33use super::engine::Events;
34use super::error::SqlxBrokerError;
35use super::publish::table_of;
36use super::queue::{Description, Timing, open};
37use super::subscriber::InboxSubscriber;
38use super::{BuiltIn, FormDialect, Plain};
39use super::{InboxRow, PayloadRow};
40
41/// A delivery of a by-name subscription whose row runs its own code, its row type erased.
42pub(crate) trait Erased: fmt::Debug + Send + Sync {
43    fn payload(&self) -> &[u8];
44    fn headers(&self) -> &HeaderMap;
45    fn decode_error(&self) -> Option<&CodecError>;
46    fn partition_key(&self) -> Option<&[u8]>;
47    fn redelivery_count(&self) -> Option<u64>;
48    fn supports_nack_after(&self) -> bool;
49    fn ack(self: Box<Self>) -> BoxFuture<'static, Result<(), AckError>>;
50    fn nack(self: Box<Self>, requeue: bool) -> BoxFuture<'static, Result<(), AckError>>;
51    fn nack_after(self: Box<Self>, delay: Duration) -> BoxFuture<'static, Result<(), AckError>>;
52}
53
54impl<DB, Row> Erased for InboxDelivery<DB, Row>
55where
56    DB: QueueDatabase,
57    Row: Events<DB> + PayloadRow,
58{
59    fn payload(&self) -> &[u8] {
60        IncomingMessage::payload(self)
61    }
62
63    fn headers(&self) -> &HeaderMap {
64        IncomingMessage::headers(self)
65    }
66
67    fn decode_error(&self) -> Option<&CodecError> {
68        IncomingMessage::decode_error(self)
69    }
70
71    fn partition_key(&self) -> Option<&[u8]> {
72        IncomingMessage::partition_key(self)
73    }
74
75    fn redelivery_count(&self) -> Option<u64> {
76        IncomingMessage::redelivery_count(self)
77    }
78
79    fn supports_nack_after(&self) -> bool {
80        IncomingMessage::supports_nack_after(self)
81    }
82
83    fn ack(self: Box<Self>) -> BoxFuture<'static, Result<(), AckError>> {
84        Box::pin(IncomingMessage::ack(*self))
85    }
86
87    fn nack(self: Box<Self>, requeue: bool) -> BoxFuture<'static, Result<(), AckError>> {
88        Box::pin(IncomingMessage::nack(*self, requeue))
89    }
90
91    fn nack_after(self: Box<Self>, delay: Duration) -> BoxFuture<'static, Result<(), AckError>> {
92        Box::pin(IncomingMessage::nack_after(*self, delay))
93    }
94}
95
96/// A delivery of a by-name subscription: a row of the table its name's route leads to.
97///
98/// It settles as an [`InboxDelivery`](crate::InboxDelivery) does. A route whose row leaves every
99/// event to the crate delivers its rows read by role: no box and no dynamic call per message, as
100/// through an [`InboxQueue`](crate::InboxQueue). A row that overrides an event, or holds a column
101/// type the crate does not read itself, settles through its own code: each delivery is boxed, and
102/// each settlement is a dynamic call whose future is boxed too.
103///
104/// # Examples
105///
106/// ```no_run
107/// # #[cfg(feature = "postgres")]
108/// # mod demo {
109/// use ruststream::OutgoingMessage;
110/// use ruststream::prelude::*;
111/// use ruststream_sqlx::{Inbox, Publish, SqlxBroker};
112/// use serde::Deserialize;
113/// use sqlx::{PgConnection, PgPool, Postgres};
114///
115/// #[derive(Inbox, sqlx::FromRow)]
116/// #[inbox(table = "email_jobs")]
117/// pub struct SendEmail {
118///     #[field(id, generated)]
119///     job_id: i64,
120///     #[field(group)]
121///     name: String,
122///     #[field(payload)]
123///     payload: Vec<u8>,
124/// }
125///
126/// impl Publish<Postgres> for SendEmail {
127///     async fn publish(
128///         conn: &mut PgConnection,
129///         message: &OutgoingMessage<'_>,
130///     ) -> Result<(), sqlx::Error> {
131///         sqlx::query("INSERT INTO email_jobs (name, payload) VALUES ($1, $2)")
132///             .bind(message.name())
133///             .bind(message.payload())
134///             .execute(conn)
135///             .await?;
136///         Ok(())
137///     }
138/// }
139///
140/// #[derive(Deserialize)]
141/// struct Email {
142///     to: String,
143/// }
144///
145/// // `emails` is a name: its route leads it to `email_jobs`, and each delivery settles there.
146/// #[subscriber("emails")]
147/// async fn send(email: &Email) -> HandlerOutcome {
148///     if email.to.is_empty() {
149///         return HandlerOutcome::drop();
150///     }
151///     HandlerOutcome::ack()
152/// }
153///
154/// pub fn app(pool: PgPool) -> RustStream {
155///     RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(
156///         SqlxBroker::new(pool).route::<SendEmail>("emails"),
157///         |b| {
158///             b.include(send);
159///         },
160///     )
161/// }
162/// # }
163/// # fn main() {}
164/// ```
165pub struct NamedDelivery<DB, D = BuiltIn<DB>>
166where
167    DB: QueueDatabase + RoleColumns,
168    D: ByName<DB> + 'static,
169{
170    delivered: Delivered<DB, D>,
171}
172
173/// A by-name delivery of either path.
174// Why the size difference stays: a row read by role is the delivery itself, and a box would cost an
175// allocation per delivery. Only `testing` widens it past the lint's bound, with the harness's
176// connection every in-process delivery carries.
177#[expect(
178    clippy::large_enum_variant,
179    reason = "a by-name delivery carries its row inline; a box would allocate per delivery"
180)]
181enum Delivered<DB, D>
182where
183    DB: QueueDatabase + RoleColumns,
184    D: ByName<DB> + 'static,
185{
186    /// A row read by role.
187    Described(InboxDelivery<DB, NamedRow<D>>),
188    /// A row read by its own code.
189    Erased {
190        delivery: Box<dyn Erased>,
191        /// Whether the delivery reports an error instead of a row, read while the row's type was
192        /// known: the box is asked for the error only where there is one.
193        refused: bool,
194    },
195}
196
197impl<DB, D> NamedDelivery<DB, D>
198where
199    DB: QueueDatabase + RoleColumns,
200    D: ByName<DB> + 'static,
201{
202    const fn described(delivery: InboxDelivery<DB, NamedRow<D>>) -> Self {
203        Self {
204            delivered: Delivered::Described(delivery),
205        }
206    }
207
208    fn erased((delivery, refused): (Box<dyn Erased>, bool)) -> Self {
209        Self {
210            delivered: Delivered::Erased { delivery, refused },
211        }
212    }
213}
214
215impl<DB, D> fmt::Debug for NamedDelivery<DB, D>
216where
217    DB: QueueDatabase + RoleColumns,
218    D: ByName<DB> + 'static,
219{
220    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
221        let mut tuple = f.debug_tuple("NamedDelivery");
222        match &self.delivered {
223            Delivered::Described(delivery) => tuple.field(delivery),
224            Delivered::Erased { delivery, .. } => tuple.field(delivery),
225        };
226        tuple.finish()
227    }
228}
229
230impl<DB, D> IncomingMessage for NamedDelivery<DB, D>
231where
232    DB: QueueDatabase + RoleColumns,
233    D: ByName<DB> + 'static,
234{
235    fn payload(&self) -> &[u8] {
236        match &self.delivered {
237            Delivered::Described(delivery) => IncomingMessage::payload(delivery),
238            Delivered::Erased { delivery, .. } => delivery.payload(),
239        }
240    }
241
242    fn headers(&self) -> &HeaderMap {
243        match &self.delivered {
244            Delivered::Described(delivery) => IncomingMessage::headers(delivery),
245            Delivered::Erased { delivery, .. } => delivery.headers(),
246        }
247    }
248
249    fn decode_error(&self) -> Option<&CodecError> {
250        match &self.delivered {
251            Delivered::Described(delivery) => IncomingMessage::decode_error(delivery),
252            Delivered::Erased {
253                delivery,
254                refused: true,
255            } => delivery.decode_error(),
256            Delivered::Erased { refused: false, .. } => None,
257        }
258    }
259
260    fn partition_key(&self) -> Option<&[u8]> {
261        match &self.delivered {
262            Delivered::Described(delivery) => IncomingMessage::partition_key(delivery),
263            Delivered::Erased { delivery, .. } => delivery.partition_key(),
264        }
265    }
266
267    fn redelivery_count(&self) -> Option<u64> {
268        match &self.delivered {
269            Delivered::Described(delivery) => IncomingMessage::redelivery_count(delivery),
270            Delivered::Erased { delivery, .. } => delivery.redelivery_count(),
271        }
272    }
273
274    async fn ack(self) -> Result<(), AckError> {
275        match self.delivered {
276            Delivered::Described(delivery) => IncomingMessage::ack(delivery).await,
277            Delivered::Erased { delivery, .. } => delivery.ack().await,
278        }
279    }
280
281    async fn nack(self, requeue: bool) -> Result<(), AckError> {
282        match self.delivered {
283            Delivered::Described(delivery) => IncomingMessage::nack(delivery, requeue).await,
284            Delivered::Erased { delivery, .. } => delivery.nack(requeue).await,
285        }
286    }
287
288    fn supports_nack_after(&self) -> bool {
289        match &self.delivered {
290            Delivered::Described(delivery) => IncomingMessage::supports_nack_after(delivery),
291            Delivered::Erased { delivery, .. } => delivery.supports_nack_after(),
292        }
293    }
294
295    async fn nack_after(self, delay: Duration) -> Result<(), AckError> {
296        match self.delivered {
297            Delivered::Described(delivery) => IncomingMessage::nack_after(delivery, delay).await,
298            Delivered::Erased { delivery, .. } => delivery.nack_after(delay).await,
299        }
300    }
301}
302
303/// The deliveries of a by-name subscription whose row runs its own code, each with whether it
304/// reports an error instead of a row.
305pub(crate) type ErasedStream =
306    Pin<Box<dyn Stream<Item = Result<(Box<dyn Erased>, bool), SqlxBrokerError>> + Send>>;
307
308/// The subscriber a by-name subscription opens: the claim loop of the table its name's route
309/// leads to.
310///
311/// It claims as an [`InboxSubscriber`](crate::InboxSubscriber) does, with the broker's poll
312/// interval, on a broker whose dialect implements [`ByName`](crate::ByName) for its database, as
313/// every built-in dialect does. A route whose row leaves every event to
314/// the crate, with role columns of types the crate reads itself, is read by those columns alone,
315/// in the row lock form and in the lease form: no box and no dynamic call per message. Each
316/// column is held to the type the struct reads it as, so a row that would not decode into the
317/// struct goes to the decode-failure policy here too. The types are an `i16`, `i32`, `i64`,
318/// `String` or `Vec<u8>` id, a `Vec<u8>` or `String` payload, headers as JSON, a text or byte key,
319/// an integer attempt, `chrono` or `time` times and leases, and
320/// [`SystemClock`](crate::SystemClock) or [`DatabaseClock`](crate::DatabaseClock). Any other row
321/// runs its own code: each delivery is boxed, each settlement is a dynamic call whose future is
322/// boxed, and each poll of the stream is a dynamic call.
323///
324/// A mount that declares `max_attempts(..)` or `dead_letter(..)` on a name is refused at startup:
325/// an [`InboxQueue`](crate::InboxQueue) descriptor takes those.
326///
327/// # Examples
328///
329/// ```
330/// # #[cfg(feature = "postgres")]
331/// # mod demo {
332/// use ruststream::OutgoingMessage;
333/// use ruststream_sqlx::prelude::*;
334/// use serde::Deserialize;
335/// use sqlx::{PgConnection, PgPool, Postgres};
336///
337/// #[derive(Inbox, sqlx::FromRow)]
338/// #[inbox(table = "report_jobs")]
339/// pub struct Report {
340///     #[field(id, generated)]
341///     id: i64,
342///     #[field(group)]
343///     name: String,
344///     #[field(payload)]
345///     payload: Vec<u8>,
346/// }
347///
348/// impl Publish<Postgres> for Report {
349///     async fn publish(
350///         conn: &mut PgConnection,
351///         message: &OutgoingMessage<'_>,
352///     ) -> Result<(), sqlx::Error> {
353///         sqlx::query("INSERT INTO report_jobs (name, payload) VALUES ($1, $2)")
354///             .bind(message.name())
355///             .bind(message.payload())
356///             .execute(conn)
357///             .await?;
358///         Ok(())
359///     }
360/// }
361///
362/// #[derive(Deserialize)]
363/// struct Request {
364///     day: String,
365/// }
366///
367/// // Each by-name mount opens a `NamedSubscriber`; the route's prefix leads both names into
368/// // `report_jobs`, and each reads its own group.
369/// #[subscriber("reports.daily")]
370/// async fn daily(request: &Request) -> HandlerOutcome {
371///     tracing::info!(day = %request.day, "daily report");
372///     HandlerOutcome::ack()
373/// }
374///
375/// #[subscriber("reports.weekly")]
376/// async fn weekly(request: &Request) -> HandlerOutcome {
377///     tracing::info!(day = %request.day, "weekly report");
378///     HandlerOutcome::ack()
379/// }
380///
381/// pub fn app(pool: PgPool) -> RustStream {
382///     RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(
383///         SqlxBroker::new(pool).route::<Report>("reports.*"),
384///         |b| {
385///             b.include(daily);
386///             b.include(weekly);
387///         },
388///     )
389/// }
390/// # }
391/// # fn main() {}
392/// ```
393pub struct NamedSubscriber<DB, D = BuiltIn<DB>>
394where
395    DB: QueueDatabase + RoleColumns,
396    D: ByName<DB> + 'static,
397{
398    opened: Opened<DB, D>,
399}
400
401/// A by-name subscription of either path.
402enum Opened<DB, D>
403where
404    DB: QueueDatabase + RoleColumns,
405    D: ByName<DB> + 'static,
406{
407    /// Rows read by role, boxed once at startup so the enum stays small.
408    Described(Box<InboxSubscriber<DB, NamedRow<D>>>),
409    /// Rows read by their own code. The stream is polled only through `&mut`, so the wrapper
410    /// shares the subscriber between threads, as a mount that publishes requires, at no cost.
411    Erased(SyncWrapper<ErasedStream>),
412}
413
414impl<DB, D> fmt::Debug for NamedSubscriber<DB, D>
415where
416    DB: QueueDatabase + RoleColumns,
417    D: ByName<DB> + 'static,
418{
419    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
420        match &self.opened {
421            Opened::Described(subscriber) => {
422                f.debug_tuple("NamedSubscriber").field(subscriber).finish()
423            }
424            Opened::Erased(_) => f.debug_struct("NamedSubscriber").finish_non_exhaustive(),
425        }
426    }
427}
428
429impl<DB, D> Subscriber for NamedSubscriber<DB, D>
430where
431    DB: QueueDatabase + RoleColumns,
432    D: ByName<DB> + 'static,
433{
434    type Message = NamedDelivery<DB, D>;
435    type Error = SqlxBrokerError;
436
437    fn stream(&mut self) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_ {
438        match &mut self.opened {
439            Opened::Described(subscriber) => Either::Left(
440                subscriber
441                    .stream()
442                    .map(|delivery| delivery.map(NamedDelivery::described)),
443            ),
444            Opened::Erased(stream) => Either::Right(
445                stream
446                    .get_mut()
447                    .as_mut()
448                    .map(|delivery| delivery.map(NamedDelivery::erased)),
449            ),
450        }
451    }
452}
453
454/// Opens the by-name subscription to `name` of `Row`'s table through `Row`'s own code, its row
455/// type erased, its statements built by `form`.
456pub(crate) fn erased<'a, DB, Row>(
457    shared: &'a Arc<Shared<DB>>,
458    form: &'a FormDialect,
459    name: &'a str,
460) -> BoxFuture<'a, Result<ErasedStream, SqlxBrokerError>>
461where
462    DB: QueueDatabase,
463    Row: InboxRow + Events<DB> + PayloadRow,
464{
465    Box::pin(async move {
466        let description = Description::of::<DB, Row>();
467        let subscriber = open::<DB, Row, Plain>(
468            shared,
469            form,
470            name,
471            Timing::default(),
472            &RetryDeclaration::new(),
473            &description,
474        )
475        .await?;
476        let stream = subscriber.into_stream().map(|delivery| {
477            delivery.map(|delivery| {
478                // Read while the row's type is known, so a delivery of a row that decoded pays no
479                // dynamic call for the runtime's question.
480                let refused = IncomingMessage::decode_error(&delivery).is_some();
481                let delivery: Box<dyn Erased> = Box::new(delivery);
482                (delivery, refused)
483            })
484        });
485        let stream: ErasedStream = Box::pin(stream);
486        Ok(stream)
487    })
488}
489
490impl<DB, D> Subscribe for ConnectedSqlxBroker<DB, D>
491where
492    DB: QueueDatabase + RoleColumns,
493    D: ByName<DB> + 'static,
494{
495    type Subscriber = NamedSubscriber<DB, D>;
496    type Copies = BrokerMoves;
497
498    async fn subscribe(&self, name: &str) -> Result<Self::Subscriber, SqlxBrokerError> {
499        let Some((route, form)) = self.shared.routes.find(name) else {
500            return Err(SqlxBrokerError::NoRoute {
501                name: name.to_owned(),
502            });
503        };
504        let description = route.description();
505        if description.kinds.is_some() {
506            tracing::debug!(
507                target: "ruststream_sqlx",
508                path = "described",
509                subscription = name,
510                table = %table_of(&description.spec),
511                row = description.row,
512                "a by-name subscription reads its rows by role",
513            );
514            let subscriber = open::<DB, NamedRow<D>, Plain>(
515                &self.shared,
516                form,
517                name,
518                Timing::default(),
519                &RetryDeclaration::new(),
520                &description.by_role(),
521            )
522            .await?;
523            return Ok(NamedSubscriber {
524                opened: Opened::Described(Box::new(subscriber)),
525            });
526        }
527        tracing::debug!(
528            target: "ruststream_sqlx",
529            path = "erased",
530            subscription = name,
531            table = %table_of(&description.spec),
532            row = description.row,
533            "a by-name subscription runs its row's own code, a box per delivery and per settlement",
534        );
535        let stream = route.subscribe(&self.shared, form, name).await?;
536        Ok(NamedSubscriber {
537            opened: Opened::Erased(SyncWrapper::new(stream)),
538        })
539    }
540}
ruststream_sqlx/inbox/named.rs

ruststream_sqlx/inbox/
named.rs
