on_threads

on_threads runs an InboxQueue subscription's handlers on dedicated threads, each with a pool of its own, counted against connection_limit.

It is the inbox's spelling of the framework's threads(n), for handlers that compute: it fills the position workers(n) fills, so the attribute names neither. InboxThreads::new(n) names the threads, by_key sends the rows of one partition key to one thread, and connections sets how many connections each thread opens, one unless set. The subscription claims on the app's runtime and runs each handler on one of its threads. Claims, settlements, transactions and publishes take the service's pool, and their connections open on the app's runtime. A handler's Ctx<keys::Pool<DB>> is the thread's own pool, with the service pool's options; its connections open on the thread and close when the thread ends. The service therefore opens its pool's connections and, for each such subscription, threads times connections. connection_limit(n) on the broker holds that sum to what the database allows: a subscription that would pass it stops the service at startup with SqlxBrokerError::ConnectionLimit. The framework's own threads(n) gives its threads pools of one connection too, and the limit does not count them. In the row lock form each delivery a thread holds keeps a connection of the service's pool, so the pool's bound applies as with workers(n) (Connections). A subscription by name takes the framework's threads(n).

Example

Four threads scan uploads, each writing its verdict through a pool of its own, and the broker holds the service to twelve connections.

Macros

examples/sqlx_subscriptions_threads.rs

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

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

/// A scan computes for a while; the pool is the thread's own.
#[subscriber(InboxQueue::<ScanJob>::new("scans"))]
async fn scan(upload: &Upload, Ctx(pool): Ctx<keys::Pool<Sqlite>>) -> HandlerOutcome {
    let verdict = if looks_clean(upload) {
        "clean"
    } else {
        "infected"
    };
    let noted = sqlx::query("INSERT INTO scan_results (path, verdict) VALUES (?, ?)")
        .bind(&upload.path)
        .bind(verdict)
        .execute(&pool)
        .await;
    if noted.is_ok() {
        HandlerOutcome::ack()
    } else {
        HandlerOutcome::retry()
    }
}

/// The service on the database `url` names, with a pool of `pool`'s size.
fn service(url: &str, pool: SqlitePoolOptions) -> RustStream {
    let schema_url = url.to_owned();
    // The service's pool and the threads' 4 x 1 connections, within a limit of 12.
    let broker = SqlxBroker::from_url(url, pool)
        .route::<ScanJob>("scans")
        .connection_limit(nonzero!(12u32));
    RustStream::new(AppInfo::new("scanner", "0.1.0"))
        // The tables, created before the subscription checks them.
        .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(scan.on_threads(InboxThreads::new(nonzero!(4))));
        })
}

Manual

examples/manual/sqlx_subscriptions_threads.rs

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

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

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

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

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

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

/// A scan computes for a while; the pool is the thread's own.
struct Scan;

impl Handle<Upload, (), (), PoolContext<Sqlite>> for Scan {
    async fn handle(
        &self,
        upload: &Upload,
        _outs: &(),
        ctx: &mut Context<'_, PoolContext<Sqlite>>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(pool) = Ctx::<keys::Pool<Sqlite>>::from_context(ctx).await?;
        let verdict = if looks_clean(upload) {
            "clean"
        } else {
            "infected"
        };
        sqlx::query("INSERT INTO scan_results (path, verdict) VALUES (?, ?)")
            .bind(&upload.path)
            .bind(verdict)
            .execute(&pool)
            .await
            .map_err(|_| HandlerOutcome::retry())?;
        Ok(())
    }
}

/// The service on the database `url` names, with a pool of `pool`'s size.
fn service(url: &str, pool: SqlitePoolOptions) -> RustStream {
    let schema_url = url.to_owned();
    // The service's pool and the threads' 4 x 1 connections, within a limit of 12.
    let broker = SqlxBroker::from_url(url, pool)
        .route::<ScanJob>("scans")
        .connection_limit(nonzero!(12u32));
    RustStream::new(AppInfo::new("scanner", "0.1.0"))
        // The tables, created before the subscription checks them.
        .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 scan = subscriber("scans", Scan)
                .map_source(|_| InboxQueue::<ScanJob>::new("scans"))
                .on_threads(InboxThreads::new(nonzero!(4)));
            b.include(scan.build());
        })
}