Skip to main content

Crate ruststream_sqlx

Crate ruststream_sqlx 

Source
Expand description

SQL databases for the RustStream messaging framework, through sqlx.

The crate has two parts. The inbox is a task queue in the service’s own table, on Postgres, MySQL, MariaDB and SQLite: a struct describes the table, SqlxBroker serves it as a RustStream broker, a subscription claims rows, and a handler’s outcome settles them. The outbox records what a service publishes on any RustStream broker until a consumer has processed it, and publishes the rest again at startup.

§A service

A queue in a Postgres table, a handler that owns its backoff, and a route that writes a publish to emails through the service’s own INSERT:

use std::time::Duration;

use chrono::{DateTime, Utc};
use ruststream::OutgoingMessage;
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
use sqlx::{PgConnection, Postgres};

// email_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT, retry_after TIMESTAMPTZ DEFAULT now(),
// attempt SMALLINT DEFAULT 1, processed_at TIMESTAMPTZ, payload BYTEA
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(retry_after, generated)]
    retry_after: DateTime<Utc>,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(processed_at, generated)]
    processed_at: Option<DateTime<Utc>>,
    #[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)]
pub struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    if deliver(email).await {
        return HandlerOutcome::ack();
    }
    // The handler owns the backoff: a minute per attempt.
    HandlerOutcome::retry_after(Duration::from_secs(60 * attempt.unwrap_or(1)))
}

#[ruststream::app]
fn app() -> impl App {
    // The options read the `PG*` environment variables.
    let broker = SqlxBroker::from_connect_options(PgConnectOptions::new(), PgPoolOptions::new());
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(
        broker.route::<SendEmail>("emails"),
        |b| {
            b.include(send);
        },
    )
}

SqlxBroker::from_connect_options only records the pool’s settings, so the app’s builder runs before the runtime starts. The runtime connects the broker at startup, which builds the pool then, and shuts it down when the service stops, which closes it. SqlxBroker::new takes a pool the service shares with code of its own and leaves it open.

§Where things are

The guides build a service on SQL databases, in this order: getting started, enqueueing jobs, retries and dead letters, long jobs and the claim form, ordering, transactional handlers, several instances, relaying events to another broker, the outbox, testing and running in production. The architecture pages show how a queue in a SQL table works and where the crate sits between the database and the runtime, and the benchmarks what it costs.

Each topic below links to its component page on the site, which shows it in a service; the item’s own documentation keeps the detail.

§Broker

§Tables

§Messages

  • Payload: payload mode, the column’s bytes lent to the codec.
  • Rows: row mode, a handler of the row itself, and RowBatch.
  • Decoding: a row that does not decode, settled by the decode policy.

§Headers

§Subscriptions

§Forms

§Settlement

  • Outcomes: what ack, retry and drop do to a row.
  • retry_after: a row hidden until its retry_after comes.
  • dead_letter: max_attempts(n) with dead_letter(..).

§Transactions

§Context

§Time

§Publishing

§Wake

§Databases

§Tooling

  • AsyncAPI: the server and the channels the generated document names (feature asyncapi).
  • Testing: the in-process mode TestApp connects the broker through (feature testing), and Live on the real clock.

§Outbox

§Cargo features

inbox turns the broker on and outbox the outbox, each independently of the other. postgres, mysql (MySQL and MariaDB) and sqlite add a built-in dialect and its sqlx driver, and any an AnyPool served by the dialect of the database it reaches. chrono or time add the time columns, and json a headers column. testing adds the broker’s in-process mode for TestApp and the outbox’s test switch, and asyncapi what a subscription adds to the generated document.

Re-exports§

pub use outbox::Outbox;
pub use outbox::OutboxDatabase;
pub use outbox::OutboxRow;
pub use outbox::OutboxSpec;
pub use outbox::OutboxTable;
pub use ruststream_sqlx_dialect as dialect;

Modules§

keys
What a handler reads off the delivery it handles, through Ctx<Key>. What a handler reads off the delivery it handles, through Ctx<Key>.
outbox
The transactional outbox: what a service publishes is recorded in its own table until a consumer has processed it, and what was not processed is published again at startup.
prelude
The one glob a service on the inbox imports: the framework’s prelude and the inbox’s own names.
spec
The settings of a queue table described by hand, as types.

Macros§

outbox
Registers outbox records under the names they track, and returns the registry.

Structs§

BuiltIn
The dialect built into the crate for the database DB: what SqlxBroker::new builds its statements with.
ClosedSqlxBroker
The broker after shutdown: what it did to the advisory locks its deliveries held.
ConnectedSqlxBroker
The connected inbox broker: the typed witness that the database answered.
DatabaseClock
The database’s own clock: #[inbox(clock = DatabaseClock)] makes the statements read it (statement_timestamp() on Postgres, UTC_TIMESTAMP(6) on MySQL) instead of binding the host’s time.
InboxDelivery
One claimed row in a handler’s hands.
InboxQueue
A subscription to a queue table: the rows of Row that the name addresses.
InboxSpec
The description of a queue table, with its settings as types.
InboxSubscriber
The subscriber an InboxQueue opens: a stream of deliveries claimed from the table.
InboxThreads
The dedicated threads of an InboxQueue subscription, and the connections each one opens for its handlers’ queries: what InboxSettings::on_threads mounts.
NamedDelivery
A delivery of a by-name subscription: a row of the table its name’s route leads to.
NamedSubscriber
The subscriber a by-name subscription opens: the claim loop of the table its name’s route leads to.
Plain
A subscription whose handler leaves the delivery’s transaction alone: the default mode.
Repository
The typed publish policy of a row with Publish: a message published through it becomes a row of Row’s table.
RepositoryPublisher
The live form of a Repository: publishes into Row’s table through its Publish.
Routed
The broker’s default publish policy: a publish goes where the route table leads its name.
RoutedPublisher
The live form of Routed: publishes into the table the route table leads each name to.
RowBatch
The rows one claim of a table in row mode took, which a batch handler borrows as one slice, &[Row], and the claims that settle them.
SqlxBroker
The inbox broker: task queues in the service’s own tables, served through a sqlx pool.
SystemClock
The host’s system clock: what a table reads unless it names another.
Transactional
A subscription whose handler writes through the delivery’s transaction, which acknowledgement commits.
Tx
The delivery’s transaction, lent to the handler: it dereferences to the connection the transaction runs on, and goes back to the delivery when the handler ends.

Enums§

NamedTime
A time a by-name subscription binds, in the type its column holds: the current time, a delayed retry’s, or in the lease form a lease’s expiry.
SqlxBrokerError
What can go wrong in the inbox broker.

Traits§

Ack
Acknowledges a row: by default it deletes the row, or sets processed_at.
AttemptColumn
The type of a field playing attempt: the delivery’s redelivery count.
AttemptRow
The row’s attempt, which a table described by hand reads off its row where its description sets InboxSpec::attempt.
BuiltInDialect
A database whose dialect is built into this crate, so SqlxBroker::new needs no dialect of the service’s own.
ByName
Subscriptions by name on the database DB, as a feature of a dialect: the headers and the times their rows hold, which no struct of the service’s own names.
Claim
Claims up to limit rows of the queue queue and returns their ids, inside the claim’s transaction.
Clock
A clock on the host: what “now” is for the statements that bind it.
DeadLetter
Moves a row whose attempts are spent to destination: by default it moves the row to that group (on a table with a group column) or into that table.
Discard
Drops a row: by default it deletes the row, or sets processed_at.
Extend
Extends the lease on a row: writes until into locked_until while the row still holds held, and says whether it did.
Fetch
Reads the rows of claimed ids, on the claim’s connection.
HeaderColumn
The type of a field playing headers: what the delivery’s header map holds.
HeaderField
A value a field of a headers struct holds, as the header the delivery’s header map carries.
HeaderFields
Header fields of the row: where its description sets InboxSpec::header_fields, the delivery builds its header map from them on its first read.
HeaderRow
A record or a row whose headers live in one column: the field holding the column, which the table’s description names with its headers setter.
InboxHeaders
A struct that describes a queue table whose message is assembled from it: the mechanics a subscription runs the queue by, and the service’s own headers. #[derive(InboxHeaders)] implements it.
InboxRow
A struct that describes a queue table; #[derive(Inbox)] implements it.
InboxSettings
The inbox’s own mount-site steps, on a #[subscriber] definition and on its settings builder.
InboxTable
A queue table described by hand.
Insert
Inserts the row: every field except the generated ones, the queue’s mechanics included, so a delayed task is a row whose retry_after lies ahead.
KeyColumn
The type of a field playing partition_key: the bytes the delivery’s key lends.
KeyRow
The row’s partition key, which a table described by hand reads off its row where its description sets InboxSpec::partition_key.
LeaseRow
A queue row taken by lease: its struct has a #[field(locked_until)] field.
Lock
Takes the advisory lock on a row’s key for the session that will hold its delivery, without waiting: the lock event of the advisory lock form.
Notifies
A database whose sessions notify each other: a broker on it may listen_notify.
PayloadRow
A queue row that carries its message as bytes: payload mode.
Publish
Writes a published message into the table: the name, the bytes and the headers reach the service’s SQL, which lays them out in its columns.
QueueDatabase
A sqlx database the inbox runs on: one that binds and reads the values the queue’s own statements use, runs statements on a connection, and reports the rows a statement changed.
QueueTime
A time a queue column holds: retry_after, processed_at or locked_until.
Retry
Releases a row for another attempt at once: by default it counts the attempt, or leaves the row for the rollback to release; in the lease form it clears the lease.
RetryAfter
Releases a row for another attempt after delay: by default it sets retry_after.
TimeColumn
The type of a field playing a time role: a QueueTime or an Option of one.
TimeSource
Where a table reads “now”: a Clock on the host, or DatabaseClock.
Unlock
Releases the advisory lock on a delivery’s key: the unlock event of the advisory lock form, which a table names as its own beside Lock.

Functions§

put_header
Puts field into headers under name, the way a field of a struct deriving InboxHeaders becomes a header; a field without a value leaves the header out.

Derive Macros§

Inbox
Describes a queue table with a struct: writes its InboxTable impl, the description a table described by hand writes, and the traits its roles need.
InboxHeaders
Describes the queue table of a message assembled from it, and implements InboxHeaders for the struct.
Outbox
Describes a service’s outbox table with a struct: implements OutboxTable, the description a record by hand writes, and HeaderRow for a headers field.