Struct SqlxBroker 

pub struct SqlxBroker<DB: Database, D = BuiltIn<DB>> { /* private fields */ }

The inbox broker: task queues in the service’s own tables, served through a sqlx pool.

new records the pool and does no I/O; the pool belongs to the service, and the broker never closes it. connect takes one connection to check the database, and picks the built-in dialect by it where the service passed none. A message in work holds one of the pool’s connections until it settles (in the lease form outside transactional mode only while it settles), and a publish takes another, so the pool is sized for both. Subscriptions read tables through InboxQueue descriptors; publishing writes them through Repository policies or through the routes this builder records.

Examples

use std::error::Error;

use ruststream::OutgoingMessage;
use ruststream::prelude::*;
use ruststream_sqlx::{Inbox, InboxQueue, Publish, SqlxBroker};
use serde::Deserialize;
use sqlx::postgres::{PgConnectOptions, PgPool};
use sqlx::{PgConnection, Postgres};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub 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)]
struct Email {
    to: String,
}

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

pub 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);
        },
    )
}

// sqlx builds a pool only inside a Tokio runtime, so the service builds it before the app.
// The service owns the pool and closes it.
#[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(())
}

Implementations

impl<DB: BuiltInDialect> SqlxBroker<DB, BuiltIn<DB>>

pub fn new(pool: Pool<DB>) -> Self

A broker on pool, with the dialect built into the crate for its database.

Synchronous and free of I/O: the database type comes from the pool, and nothing connects before connect, which picks the dialect from the connection it checks. An AnyPool reaches a database named only then, and a database whose dialect’s feature is off stops connect with SqlxBrokerError::Backend.

Examples

use std::error::Error;

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::postgres::{PgConnectOptions, PgPool, PgPoolOptions};

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

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

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

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

#[tokio::main]
pub async fn main() -> Result<(), Box<dyn Error>> {
    // Nothing connects here: the pool connects lazily, and the broker once the service
    // starts. sqlx builds even a lazy pool only inside a Tokio runtime, so it is built
    // here rather than in the app's builder. The options read the `PG*` environment
    // variables.
    let pool = PgPoolOptions::new().connect_lazy_with(PgConnectOptions::new());
    let app = app(pool);
    app.run().await?;
    Ok(())
}

impl<DB: QueueDatabase, D: Dialect + 'static> SqlxBroker<DB, D>

pub fn with_dialect(pool: Pool<DB>, dialect: D) -> Self

A broker on pool whose statements dialect, a dialect of the service’s own, builds: a driver this crate has no dialect for, or a built-in dialect the service wraps to write a statement its own way.

The dialect’s type is the broker’s second type parameter, so what a subscription asks of it is checked where the subscription mounts: a table in the row lock form needs the dialect to implement RowLock, a lease table Lease, and a table that names an isolation level or a SQLite mode Opens for it. The dialect runs while subscriptions start; messages travel through the statements it built.

Examples

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

/// Postgres with statements of the service's own, written out in the crate overview.
#[derive(Debug)]
pub struct Audited;

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

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

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

pub fn app(pool: PgPool) -> RustStream {
    // A `SqlxBroker<Postgres, Audited>`: `send` mounts only because `Audited` implements
    // `RowLock`, the form of `email_jobs`.
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        .with_broker(SqlxBroker::with_dialect(pool, Audited), |b| {
            b.include(send);
        })
}

pub fn route<Row>(self, name: impl Into<Cow<'static, str>>) -> Self

where Row: Publish<DB> + Events<DB> + PayloadRow, Row::Form: FormOn<D>, D: Opens<Row::Opening>,

Leads publishes to name into the table of Row, through its Publish, and opens a subscription by that name (#[subscriber("name")]) on the same table.

A name ending in * leads every name that starts with what precedes it; an exact name wins over a prefix, and a longer prefix over a shorter one. Routing a name twice leads it to the last row type. The broker’s default publisher is this route table: a reply or an Out slot without a policy of its own publishes through it.

A route asks of the broker’s dialect what an InboxQueue of the row asks: the table’s form, and the isolation level or SQLite mode the struct declares. A by-name subscription names no row type, so the route is where its table is checked.

Examples

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

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

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

#[derive(Deserialize)]
struct Daily {
    day: u32,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "reports.weekly")]
struct Weekly {
    week: u32,
}

// Read by name from `report_jobs`, answered with a row of `report_jobs` in group `reports.weekly`.
#[subscriber("reports.daily", reply)]
async fn roll_up(daily: &Daily) -> Weekly {
    Weekly { week: daily.day / 7 }
}

pub fn app(pool: PgPool) -> RustStream {
    // `reports.daily` and `reports.weekly` both lead into `report_jobs`.
    let broker = SqlxBroker::new(pool).route::<Report>("reports.*");
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(broker, |b| {
        b.include(roll_up);
    })
}

pub const fn poll_interval(self, interval: Duration) -> Self

How long a subscription waits between claims that found its queue empty; one second unless set. A subscription’s own poll_interval overrides it. A publish through this broker wakes the subscriptions of its table and group before then, and so does a notification with listen_notify (waking a subscription).

Examples

use std::time::Duration;

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

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

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

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

pub fn app(pool: PgPool) -> RustStream {
    // Every queue on this broker checks twice a second when it runs dry.
    let broker = SqlxBroker::new(pool).poll_interval(Duration::from_millis(500));
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
        b.include(send);
    })
}

pub const fn lease(self, lease: Duration) -> Self

How long a claim leases a row of a table in the lease form (one with a #[field(locked_until)] field); thirty seconds unless set. A subscription’s own lease overrides it, and a table in another form ignores it.

A lease is whole seconds, at least one: a shorter one is rounded up. The subscription extends the lease of every delivery in work each half lease, so a handler may run longer than its lease; the lease is how long a row stays out of the queue after its process crashed. A delivery dropped unsettled releases its row at once. A lease runs out under a running handler only when its extensions fail or stop (the database out of reach, the subscription closed, the broker shut down): the row then goes to the next claim, and the late settlement fails with SqlxBrokerError::LeaseLost.

Examples

use std::time::Duration;

use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "video_jobs")]
pub struct Transcode {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Video {
    path: String,
}

#[subscriber(InboxQueue::<Transcode>::new("videos"))]
async fn transcode(video: &Video) -> HandlerOutcome {
    tracing::info!(path = %video.path, "transcoding");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    // A video whose process crashed goes back to the queue once its ten-second lease runs out.
    let broker = SqlxBroker::new(pool).lease(Duration::from_secs(10));
    RustStream::new(AppInfo::new("transcoder", "0.1.0")).with_broker(broker, |b| {
        b.include(transcode);
    })
}

pub const fn connection_limit(self, limit: NonZeroU32) -> Self

The most connections the service opens on the database: the pool’s size, and the connections of the dedicated threads of every subscription mounted with on_threads. No limit unless set.

Each such subscription counts its threads times their connections when it opens. One that would take the sum past the limit refuses to open, with SqlxBrokerError::ConnectionLimit naming the numbers, and the service does not start. Set it to what the database lets this service open.

Examples

use ruststream::nonzero;
use ruststream_sqlx::prelude::*;
use sqlx::{PgPool, Postgres};

#[subscriber(InboxQueue::<ScanJob>::new("scans"))]
async fn scan(image: &Image, Ctx(pool): Ctx<keys::Pool<Postgres>>) -> HandlerOutcome {
    let noted = sqlx::query("INSERT INTO scanned (path) VALUES ($1)")
        .bind(&image.path)
        .execute(&pool)
        .await;
    if noted.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
}

pub fn app(pool: PgPool) -> RustStream {
    // The database lets this service open 20 connections: the pool's and 4 threads' own.
    let broker = SqlxBroker::new(pool).connection_limit(nonzero!(20u32));
    RustStream::new(AppInfo::new("scanner", "0.1.0")).with_broker(broker, |b| {
        b.include(scan.on_threads(InboxThreads::new(nonzero!(4))));
    })
}

pub fn listen_notify(self) -> Self

where DB: Notifies,

Wakes the broker’s subscriptions on Postgres notifications: a row another process announced with pg_notify is claimed at once, not on the next poll.

The broker holds one connection of the pool for its life, from connect to shutdown, and listens there. Each subscription listens on the channel of its table, qualified with its schema, as it opens; a notification whose payload names a group wakes that group’s subscription, and one with an empty payload wakes every subscription of the table. Each publish of the broker sends SELECT pg_notify(table, group) on the connection that wrote the row, right after the write: one more statement per publish. A publish whose notification fails is still a publish; the row waits for the poll interval and the failure is logged.

The poll interval stays: a notification sent while the listening connection was lost is gone, so after a reconnect every subscription claims once, and a row written by the service’s own SQL or a handler’s Tx wakes nobody unless the service sends the notification itself. It is opt-in because of its cost: the pool connection it holds, the statement each publish adds, and the lock, global to the server, that Postgres takes at the commit of each transaction that notifies, so those commits run one after another, which limits many concurrent writers.

A table whose qualified name is longer than Postgres’s 63-byte channel names stops its subscription at startup with SqlxBrokerError::Declaration.

Examples

use std::time::Duration;

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

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

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

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

pub fn app(pool: PgPool) -> RustStream {
    // A row another service writes and announces with
    // `SELECT pg_notify('email_jobs', 'emails')` is sent at once; the poll each minute
    // catches a row whose notification was lost.
    let broker = SqlxBroker::new(pool)
        .poll_interval(Duration::from_secs(60))
        .listen_notify();
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(broker, |b| {
        b.include(send);
    })
}

Trait Implementations

impl<DB: QueueDatabase, D: Dialect + 'static> Broker for SqlxBroker<DB, D>

type Error = SqlxBrokerError

The error type returned by broker-level operations.

type Connected = ConnectedSqlxBroker<DB, D>

The connected form of this broker: the typed witness that connect succeeded.

async fn connect(self) -> Result<Self::Connected, Self::Error>

Establishes the connection to the broker, consuming the unconnected form.

fn bindable(self) -> Bindable<Self>

where Self: 'static,

Wraps the broker for publisher-token minting before registration: the returned Bindable hands out Bound tokens via its bind, and is itself what with_broker takes.

impl<DB: Database, D: Dialect> Debug for SqlxBroker<DB, D>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<D: Dialect + 'static> DescribeServer for SqlxBroker<Postgres, D>

Available on crate feature postgres only.

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl<D: Dialect + 'static> DescribeServer for SqlxBroker<MySql, D>

Available on crate feature mysql only.

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl<D: Dialect + 'static> DescribeServer for SqlxBroker<Sqlite, D>

Available on crate feature sqlite only.

SQLite runs inside the service, on a file or in memory: the description names the protocol and no host, so it holds neither the database’s path nor any other part of its URL, and SQLite takes no credentials.

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl<D: Dialect + 'static> DescribeServer for SqlxBroker<Any, D>

Available on crate feature any only.

An AnyPool is described by the scheme of its URL, as the broker of that database describes itself: the host and port of a server, or SQLite with no host.

fn describe_server(&self) -> ServerSpec

Returns the server coordinates for this broker.

impl<DB: QueueDatabase, D: Dialect + 'static> InProcess for SqlxBroker<DB, D>

async fn connect_in_process( self, ) -> Result<ConnectedSqlxBroker<DB, D>, SqlxBrokerError>

Connects to the database the test’s pool reaches, as connect does, with every database call of the connection kept off a paused clock.