SqlxBroker

SqlxBroker serves the queues in the tables of the service's database through a sqlx pool; it builds the pool when it connects, or takes the service's own.

SqlxBroker::from_url records the database's URL and the pool's options, and from_connect_options records the driver's connect options in place of the URL. Neither does I/O, so the service builds the broker in its #[ruststream::app] builder, before the runtime starts. connect builds the pool, and shutdown closes it. SqlxBroker::new takes a Pool<DB> the service owns and shares with the rest of its code; the broker never closes that pool. sqlx builds a pool only inside a Tokio runtime, so a service that owns its pool builds it in its own main. The database is one a built-in dialect serves: Postgres, MySQL, MariaDB or SQLite. The settings that follow the constructor are builder steps: routes that lead publishes into tables, the poll interval and the lease every subscription takes unless it names its own, and the connection limit of dedicated threads.

At startup the runtime connects the broker and receives a ConnectedSqlxBroker. The connect takes one connection from the pool, so a database out of reach fails startup with SqlxBrokerError::Connect, and a URL the driver cannot read with SqlxBrokerError::Url. It also picks the dialect from that connection, which matters for an AnyPool. Every subscription then reads one queue through an InboxQueue descriptor, and the startup checks run as it opens. At shutdown the runtime turns the connected broker into a ClosedSqlxBroker.

A message in work holds connections of the pool by the form its table claims rows in, and a publish takes one more; Connections shows how to size the pool.

Example

The service serves the queue thumbnails from a SQLite file: the broker builds its pool from the file's URL, and on_startup creates the table before the subscription checks it. A test hands service a database of its own. cargo run -p ruststream-sqlx --example sqlx_broker --features inbox,sqlite,chrono,testing starts it.

Macros
Manual

examples/sqlx_broker.rs
use ruststream_sqlx::prelude::*;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Connection, Sqlite, SqliteConnection};

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

#[subscriber(InboxQueue::<MakeThumbnail>::new("thumbnails"))]
async fn thumbnail(image: &Image) -> HandlerOutcome {
    tracing::info!(path = %image.path, "making a thumbnail");
    HandlerOutcome::ack()
}

/// 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("thumbnails", "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.route::<MakeThumbnail>("thumbnails"), |b| {
            b.include(thumbnail);
        })
}

#[ruststream::app]
fn app() -> impl App {
    service("sqlite:thumbnails.db?mode=rwc")
}

examples/manual/sqlx_broker.rs
use ruststream_sqlx::prelude::*;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Connection, Sqlite, SqliteConnection};

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

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

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

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

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

struct Thumbnail;

impl Handle<Image> for Thumbnail {
    fn handle(
        &self,
        image: &Image,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(path = %image.path, "making a thumbnail");
        ready(Ok(()))
    }
}

/// 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("thumbnails", "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.route::<MakeThumbnail>("thumbnails"), |b| {
            let thumbnail = subscriber("thumbnails", Thumbnail)
                .map_source(|_| InboxQueue::<MakeThumbnail>::new("thumbnails"));
            b.include(thumbnail.build());
        })
}

fn main() -> ExitCode {
    run_main(|| service("sqlite:thumbnails.db?mode=rwc"))
}

Example​

Example