Batches

A slice handler gets one claim of up to batch(n) rows; each row settles on its own, and in the row lock form the settlements commit together.

A handler over &[T] is a batch handler, and batch(n) at the mount site is the most rows one claim takes for it. A claim takes what is due at once and waits for nothing, so a quiet queue yields batches smaller than n. A Vec<HandlerOutcome> settles row i with outcome i, and a single outcome settles them all. A row that does not decode is settled by the decode policy, and the batch keeps its other rows. On SQLite the rows of a batch come in no particular order. Each form settles a batch its own way:

Form	A batch
row lock	shares the claim's transaction and one connection; each row settles in it, and the settlements take effect together when the last delivery finishes
lease	settles each row on its own, on a connection taken for the statement
advisory lock	holds a connection per row and settles each row on it; it holds one row per key and ends where the pool has no connection to spare

In the row lock form a settlement whose statement fails rolls the whole batch back, its rows return, and the later settlements fail with SqlxBrokerError::BatchRolledBack. A table in row mode hands a batch handler &[Row], the rows themselves (Rows). A batch handler runs outside transactional mode: a batch mounted with .transactional() does not compile. In the manual form the body is an impl Handle<[T]>, where Ok(()) acknowledges the batch and Err(outcomes) settles it row by row.

Example

One claim takes up to fifty emails, and an address that cannot receive mail drops its own row alone.

Macros

examples/sqlx_subscriptions_batches.rs

use ruststream_sqlx::prelude::*;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Connection, SqliteConnection};

/// One row of `newsletter_jobs`, claimed under a lease.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "newsletter_jobs")]
struct SendNewsletter {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// Up to fifty emails per claim; an address that cannot receive mail is dropped, and only its
/// own row is.
#[subscriber(InboxQueue::<SendNewsletter>::new("newsletter"))]
async fn send_all(emails: &[Email]) -> Vec<HandlerOutcome> {
    tracing::info!(count = emails.len(), "sending a batch");
    emails
        .iter()
        .map(|email| {
            if email.to.contains('@') {
                HandlerOutcome::ack()
            } else {
                HandlerOutcome::drop()
            }
        })
        .collect()
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("newsletter", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            b.include(send_all.batch(nonzero!(50)));
        })
}

Manual

examples/manual/sqlx_subscriptions_batches.rs

use ruststream_sqlx::prelude::*;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Connection, SqliteConnection};

/// One row of `newsletter_jobs`, claimed under a lease.
#[derive(sqlx::FromRow)]
struct SendNewsletter {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for SendNewsletter {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("newsletter_jobs", Column::new("id").generated())
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

    fn id(&self) -> &i64 {
        &self.id
    }
}

impl PayloadRow for SendNewsletter {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

/// Up to fifty emails per claim; an address that cannot receive mail is dropped, and only its
/// own row is.
struct SendAll;

impl Handle<[Email]> for SendAll {
    fn handle(
        &self,
        emails: &[Email],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = emails.len(), "sending a batch");
        let outcomes = emails
            .iter()
            .map(|email| {
                if email.to.contains('@') {
                    HandlerOutcome::ack()
                } else {
                    HandlerOutcome::drop()
                }
            })
            .collect();
        // One outcome per email, in the batch's order.
        ready(Err(outcomes))
    }
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("newsletter", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            let send_all = subscriber("newsletter", SendAll)
                .map_source(|_| InboxQueue::<SendNewsletter>::new("newsletter"))
                .batch(nonzero!(50));
            b.include(send_all.build());
        })
}