Databases

SqlxBroker takes the dialect built in for its database; statements are built once when a subscription opens, the insert at compile time.

A dialect builds the SQL of a broker's tables from their descriptions. The built-in one, BuiltIn<DB>, is the broker's default second type parameter: SqlxBroker<Sqlite> and SqlxBroker<Sqlite, BuiltIn<Sqlite>> name the same type. Each database has a feature that adds its dialect and its sqlx driver: postgres, mysql for MySQL and MariaDB, and sqlite. any serves an AnyPool by the built-in dialect of the database it reaches. A database without a built-in dialect is served by a dialect of the service's own.

One struct describes a queue table on every database. The broker takes the dialect of its database, which builds the claims and settlements of each subscription once, when it opens, so a message costs the same on every database. The derive renders the insert of every enabled dialect at compile time. The service writes its Publish per database, because that statement is its own SQL.

What each database serves:

	Postgres	MySQL, MariaDB	SQLite	`AnyPool`
Row lock form	yes	yes	no	Postgres and MySQL backends
Lease form	yes	yes	yes	no
Advisory lock form	yes	yes	in the process	yes
Isolation levels	three	four	three modes	the backend's
`listen_notify`	yes	no	no	no
Database clock	`statement_timestamp()`	`UTC_TIMESTAMP(6)`	`strftime(..)`, as text	no time columns

The connections a subscription holds depend on its form and on transactional mode, alike on every database: Connections shows how to size the pool.

Example

The broker's type names the built-in SQLite dialect, the default of SqlxBroker::from_url.

Macros
Manual

examples/sqlx_databases.rs
use ruststream_sqlx::BuiltIn;

/// One row of `thumbnail_jobs`, claimed under a lease. The same struct serves a `PgPool` and a
/// `MySqlPool`: only its `Publish` is written per database.
#[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()
}

// `SqlxBroker<Sqlite>` names the same type: the built-in dialect of the database is the default.
fn broker(url: &str) -> SqlxBroker<Sqlite, BuiltIn<Sqlite>> {
    SqlxBroker::from_url(url, SqlitePoolOptions::new()).route::<MakeThumbnail>("thumbnails")
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    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(url), |b| {
            b.include(thumbnail);
        })
}

examples/manual/sqlx_databases.rs
use ruststream_sqlx::BuiltIn;

/// One row of `thumbnail_jobs`, claimed under a lease. The same description serves a `PgPool`
/// and a `MySqlPool`: only its `Publish` is written per database.
#[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(()))
    }
}

// `SqlxBroker<Sqlite>` names the same type: the built-in dialect of the database is the default.
fn broker(url: &str) -> SqlxBroker<Sqlite, BuiltIn<Sqlite>> {
    SqlxBroker::from_url(url, SqlitePoolOptions::new()).route::<MakeThumbnail>("thumbnails")
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    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(url), |b| {
            let thumbnail = subscriber("thumbnails", Thumbnail)
                .map_source(|_| InboxQueue::<MakeThumbnail>::new("thumbnails"));
            b.include(thumbnail.build());
        })
}

Example​

Example