Struct RowBatch 

pub struct RowBatch<DB, Row>
where
    DB: QueueDatabase,
    Row: Events<DB> + QueueRow<Lane = RowLane>,
{ /* private fields */ }

The rows one claim of a table in row mode took, which a batch handler borrows as one slice, &[Row], and the claims that settle them.

The rows lie in one vector in claim order, as the driver read them: the handler reads them where they lie, with no codec and no copy. Each row’s headers column was taken into its delivery’s headers when the batch was built, so the handler reads the column empty. A message assembled from a headers struct keeps its headers struct whole, and each delivery builds its header map from it on the first read. A claimed id without a row, and a row the driver could not read, follow the slice: the runtime settles them by the subscription’s decode policy, its error in the log. Every delivery of the batch holds its row in the table’s form, as a payload-mode batch’s deliveries do: one transaction for the whole batch in the row lock form, a lease the subscription extends while the handler runs in the lease form, a session that holds the row’s key in the advisory lock form.

A batch dropped before its deliveries settle (its handler panicked, the app shut down) returns its rows at once, as a dropped delivery does.

Examples

use ruststream_sqlx::prelude::*;
use sqlx::PgPool;

/// A mail to send: the table has no payload column, so a handler takes the rows themselves.
#[derive(Debug, Clone, Inbox, sqlx::FromRow)]
#[inbox(table = "mail_jobs")]
pub struct SendEmail {
    #[field(id)]
    job_id: i64,
    recipient: String,
}

/// Each claim of up to 50 rows reaches the handler as one slice; each mail settles on its own.
#[subscriber(InboxQueue::<SendEmail>::new("mail"))]
async fn send(mails: &[SendEmail]) -> Vec<HandlerOutcome> {
    let mut outcomes = Vec::with_capacity(mails.len());
    for mail in mails {
        outcomes.push(if deliver(&mail.recipient).await {
            HandlerOutcome::ack()
        } else {
            HandlerOutcome::retry()
        });
    }
    outcomes
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(send.batch(nonzero!(50)));
    })
}

Trait Implementations

impl<DB, Row> CarriesBatch<Row> for RowBatch<DB, Row>

where DB: QueueDatabase, Row: Events<DB> + QueueRow<Lane = RowLane>,

fn carried(&self) -> &[Row]

The values the batch’s deliveries carry, in the order the batch yields its deliveries.

impl<DB, Row> Debug for RowBatch<DB, Row>

where DB: QueueDatabase, Row: Events<DB> + QueueRow<Lane = RowLane>,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<DB, Row> IntoIterator for RowBatch<DB, Row>

where DB: QueueDatabase, Row: Events<DB> + QueueRow<Lane = RowLane>,

type Item = InboxDelivery<DB, Row>

The type of the elements being iterated over.

type IntoIter = RowDeliveries<DB, Row>

Which kind of iterator are we turning this into?

fn into_iter(self) -> Self::IntoIter

Creates an iterator from a value.