Skip to main content

Getting started

This guide installs ruststream-sqlx, creates a queue table and runs a first service on it, with SQLite and no server.

Installation​

Cargo.toml
[dependencies]
ruststream = { version = "0.7", features = ["macros", "json", "logging"] }
ruststream-sqlx = { version = "0.7", features = ["inbox", "sqlite", "chrono"] }
sqlx = { version = "0.9", features = ["runtime-tokio", "sqlite", "derive", "chrono"] }
chrono = "0.4"
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
tracing = "0.1"

[dev-dependencies]
ruststream-sqlx = { version = "0.7", features = ["inbox", "sqlite", "chrono", "testing"] }

ruststream is the framework: macros gives the attributes, json the default codec, and logging prints the service's log to the terminal. ruststream-sqlx serves the queue, and sqlx is the client the service already uses for its database. The testing feature adds the in-process mode that tests run on.

The crate has no default features. A service turns on what it uses:

FeatureWhat it gives
inboxthe task queue in the service's tables, served by SqlxBroker
outboxthe transactional outbox over any RustStream broker
postgres, mysql, sqlitea built-in dialect and its sqlx driver; mysql serves MySQL and MariaDB
anyan AnyPool, served by the dialect of the database it reaches
chrono, timethe time columns in that crate's types
jsona JSON headers column
testingthe in-process mode behind TestApp, and the outbox's test switch
asyncapithe tables in the generated AsyncAPI document

The table​

The queue is a table of the service's own, and the service creates it:

src/main.rs
const SCHEMA: &str = "CREATE TABLE IF NOT EXISTS thumbnail_jobs (
    id INTEGER PRIMARY KEY,
    attempt INTEGER NOT NULL DEFAULT 1,
    locked_until TEXT,
    payload BLOB NOT NULL
)";

Each row is one job. id names the row, attempt counts its deliveries, and payload holds the encoded message. locked_until holds the lease of a row in work: SQLite has no row locks, so a SQLite queue claims its rows under a lease. Long jobs, crashes and the claim form compares the forms a table can take.

The first service​

src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

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

/// A publish to `thumbnails` writes a row with the service's own statement.
impl Publish<Sqlite> for MakeThumbnail {
    async fn publish(
        conn: &mut SqliteConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO thumbnail_jobs (payload) VALUES (?)")
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

/// The job, as the payload column holds it.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "thumbnails")]
struct Image {
    path: String,
}

#[subscriber(InboxQueue::<MakeThumbnail>::new("thumbnails"))]
async fn thumbnail(image: &Image, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    if image.path.is_empty() {
        tracing::warn!("an image without a path, dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(path = %image.path, ?attempt, "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);
        })
}

#[derive(Inbox)] describes the table: #[inbox(table = ..)] names it, and each #[field(..)] gives a column its role. generated marks a column the database fills. In the manual form, InboxSpec says the same. InboxQueue::<MakeThumbnail>::new("thumbnails") is the subscription: it claims rows of thumbnail_jobs and hands each payload to the codec. What the handler returns settles the row: ack() deletes it, and drop() finishes it the same way without a thumbnail. Ctx<keys::Attempt> reads which delivery of the row this is.

service takes the database's URL. SqlxBroker::from_url records it with the pool's options and connects nothing, so the app is built before the runtime starts. on_startup creates the table before the subscription checks it.

A publish to thumbnails writes a row. route leads the name to MakeThumbnail, and the service's own Publish statement inserts it. A handler's reply or a test publishes the same way. Code outside a handler, such as an HTTP endpoint, writes the row with the derive's insert, inside its own transaction: Enqueueing jobs shows it.

Running the service​

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

#[ruststream::app] generates main, which builds the app and runs it. The broker builds its pool when the service starts, and closes it when the service stops.

cargo run

The service claims rows as they appear in thumbnails.db. A row another program inserts is seen within the poll interval, one second unless set. Ctrl+C stops it.

The first test​

The example file also holds two tests. Each opens a SQLite database in memory and starts the same service on it with TestApp::start. A test publishes an image through the route, as production does, and checks how thumbnail settled it and what is left in the table:

cargo test

Testing a service on SQL walks through these tests.

Moving to Postgres or MySQL​

The same struct serves a PgPool or a MySqlPool: the broker takes the dialect of its pool. Turn on postgres or mysql in both crates, build that pool, and write Publish<Postgres> or Publish<MySql> with the database's placeholders. On a server, a table without locked_until claims its rows under a row lock, held while the handler runs. Index the claim order, (group, priority, retry_after, id) for the columns the table has; Running in production explains why.

Coming from a hand-written queue​

A queue on sqlx usually is a loop of SELECT .. FOR UPDATE SKIP LOCKED and a few columns of bookkeeping. Here is where each part goes:

Hand-writtenruststream-sqlx
the claim loop and its statementan InboxQueue subscription; the dialect builds the statement
DELETE or UPDATE .. SET done after the workthe handler's outcome
an attempts column raised by hand#[field(attempt)] and Ctx<keys::Attempt>
a run_at column for later#[field(retry_after)] and retry_after(d) (Retries)
a locked_until column and a reaperthe lease form (Long jobs)
a dead-letter table and its movemax_attempts(n) and dead_letter
a job inserted with the data it servesinsert in the same transaction (Enqueueing jobs)