ruststream-sqlx 0.7.0

SQL databases for the RustStream messaging framework through sqlx: a transactional outbox over any RustStream broker and task queues in Postgres, MySQL/MariaDB and SQLite tables.
Documentation

ruststream-sqlx connects a RustStream service to SQL databases over sqlx. Handlers, routing, codecs and middleware come from the framework; this crate brings two components: the inbox, a broker that serves task queues from the service's own tables, and the transactional outbox, which tracks what a service publishes on any RustStream broker.

Features

  • Task queues in the service's own tables on Postgres, MySQL 8.0.1 and later, MariaDB 10.6 and later, and SQLite, or through an AnyPool.
  • Three claim forms: a row lock held in a transaction while the handler runs, a lease committed at once and extended while the handler works, or an advisory lock on the row's key. SQLite tables take the lease or the advisory lock form.
  • Transactional mode: .transactional() lets a handler write through its delivery's transaction, and the acknowledgement commits those writes together with the row.
  • Column roles that each turn on one behaviour: groups, FIFO keys, delayed retries, attempt caps with dead letters, a processed mark, and the database's clock in place of the host's.
  • Row mode, headers layouts and batches: a handler takes the decoded payload, the row itself, a struct joined from other tables, or the rows of one claim as a slice.
  • A derive or the manual API: #[derive(Inbox)] or a trait with a typed builder, checked as strictly by the compiler and running the same statements.
  • SQL checked at compile time: checked hands the derive's statements to sqlx's macros, so cargo sqlx prepare checks them with the service's own queries.
  • The service's own SQL for any queue event, and a dialect of its own for another database.
  • Waking on publish: a publish wakes the subscriptions of its table in the same process, and LISTEN/NOTIFY on Postgres wakes them from other processes.
  • The transactional outbox over any RustStream broker: #[derive(Outbox)] describes the service's outbox table, and outbox! registers it under the names it tracks. A publish middleware records each tracked message before it is sent, with the record's id in a header. A subscription middleware takes the record into work by that id and marks it processed when the handler acknowledges. Unprocessed records are published again at startup, so each message is delivered at least once.
  • Tests on the production app: TestApp runs the service's own app against a real database; SQLite needs no server.

Inbox or outbox

The inbox fits work that belongs to the service's data: a task written in the same transaction as the data it serves, a queue in the database the service already runs. The outbox fits messages a service publishes on another broker and must not lose. A service may use both: the outbox's middlewares wrap inbox handlers as they wrap any other.

Install

[dependencies]
ruststream = { version = "0.7", features = ["macros", "json"] }
ruststream-sqlx = { version = "0.7", features = ["inbox", "outbox", "postgres"] }
sqlx = { version = "0.9", features = ["runtime-tokio", "postgres", "derive", "chrono"] }
chrono = "0.4"
serde = { version = "1", features = ["derive"] }
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

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

inbox and outbox turn the components on, each independently of the other. A driver feature picks the database: postgres, mysql (MySQL and MariaDB), sqlite, or any. Optional features: chrono or time for time columns, json for a headers column, and asyncapi.

Write a service

A queue in a Postgres table, served by the inbox:

use std::error::Error;

use ruststream::OutgoingMessage;
use ruststream_sqlx::prelude::*;
use serde::{Deserialize, Serialize};
use sqlx::postgres::{PgConnectOptions, PgConnection, PgPool, Postgres};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

impl Publish<Postgres> for SendEmail {
    async fn publish(
        conn: &mut PgConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO email_jobs (name, payload) VALUES ($1, $2)")
            .bind(message.name())
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[derive(Deserialize, Serialize, Outgoing)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    println!("sending to {}", email.to);
    HandlerOutcome::ack()
}

fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(
        SqlxBroker::new(pool).route::<SendEmail>("emails"),
        |b| {
            b.include(send);
        },
    )
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let pool = PgPool::connect_with(PgConnectOptions::new()).await?;
    app(pool.clone()).run().await?;
    pool.close().await;
    Ok(())
}

sqlx builds a pool only inside a Tokio runtime, so main builds it and hands it to the app.

Messages published on any broker, tracked by the outbox:

use std::convert::Infallible;

use chrono::{DateTime, Utc};
use ruststream::OutgoingMessage;
use ruststream::memory::prelude::*;
use ruststream_sqlx::{Outbox, outbox};
use serde::{Deserialize, Serialize};
use sqlx::postgres::{PgConnectOptions, PgConnection, PgPool, Postgres};

#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "outbox")]
struct OrderOutbox {
    #[field(id)]
    id: i64,
    #[field(name)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
}

impl outbox::Publish<Postgres> for OrderOutbox {
    async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
        sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
            .bind(msg.name())
            .bind(msg.payload())
            .fetch_one(conn)
            .await
    }
}

#[derive(Deserialize)]
struct PlaceOrder {
    id: u64,
}

#[derive(Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct OrderPlaced {
    id: u64,
}

#[subscriber("checkout", reply)]
async fn place(cmd: &PlaceOrder) -> OrderPlaced {
    OrderPlaced { id: cmd.id }
}

#[subscriber("orders")]
async fn fulfil(order: &OrderPlaced) -> HandlerOutcome {
    println!("fulfilling order {}", order.id);
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let tracking = outbox! {
        "orders" => OrderOutbox,
    };
    let registry = tracking.clone();

    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .on_startup(async move |()| {
            let pool = PgPool::connect_with(PgConnectOptions::new()).await?;
            registry
                .set_pool(pool.clone())
                .expect("the outbox is deferred and gets one pool");
            Ok::<_, sqlx::Error>(pool)
        })
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        .after_shutdown(async move |pool| {
            pool.close().await;
            Ok::<_, Infallible>(())
        })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            b.include(fulfil);
            b.after_startup(Publish, tracking.republish());
        })
}

#[ruststream::app] generates main and builds the app before the runtime starts, so the outbox starts without a pool and on_startup hands it one with set_pool. Both services are in crates/ruststream-sqlx/examples, and examples/outbox runs the outbox over Redis Pub/Sub.

Test it

TestApp runs the service's own app against a real database, because the service's SQL is part of what the test checks.

use ruststream::testing::TestApp;

let pool = PgPool::connect(&std::env::var("DATABASE_URL")?).await?;
let tb = TestApp::start_live(app(pool)).await?;

tb.broker::<SqlxBroker<Postgres>>()
    .message(&Email { to: "ann@example.com".to_owned() })
    .to("emails")
    .publish()
    .await?;

tb.broker::<SqlxBroker<Postgres>>()
    .subscriber("emails")
    .assert_called_once()
    .settled(HandlerOutcome::ack());

A test build leaves the outbox off, so a test needs no outbox table; RUSTSTREAM_SQLX_OUTBOX=on in the environment turns it on.

Documentation

Minimum supported Rust version

The MSRV is 1.95, edition 2024.

Contributing

See CONTRIBUTING.md.

License

Licensed under the Apache-2.0 license.