AnyPool

With the any feature, an AnyPool is served by the built-in dialect of the database it reaches at connect; its rows hold integers, text and bytes only.

An AnyPool names its database by the scheme of its URL, so the service picks the database in its configuration and builds one binary for all of them. sqlx::any::install_default_drivers() registers the drivers before the pool connects. The broker picks the dialect when it connects, among the dialects whose features are on; another backend stops connect with SqlxBrokerError::Backend.

A row holds the types sqlx::Any carries: integers, text and bytes, with no time and no JSON. A table on an AnyPool therefore has no locked_until, retry_after, processed_at or headers column, and the lease form is out of its reach. It serves the row lock form on its Postgres and MySQL backends and the advisory lock form on all three. A row lock table on a SQLite backend stops its subscription at startup, because the database is known only then. For the same reason a table's isolation level compiles at every level, and the backend's dialect refuses one its database does not open when the subscription starts, with SqlxBrokerError::Dialect.

One Publish serves every backend when it writes the row through the derive's Insert on the connection the pool hands out. The AsyncAPI document describes the server by the URL's scheme, never its credentials. A test addresses the broker as tb.broker::<SqlxBroker<Any>>().

Example

An email queue in the advisory lock form, which every backend serves; the test runs it on SQLite in memory.

Macros

examples/sqlx_databases_any.rs

use ruststream::OutgoingMessage;
use ruststream_sqlx::QueueDatabase;
use sqlx::any::AnyPoolOptions;
use sqlx::{Any, AnyConnection, Connection};

/// One row of `email_jobs` in the advisory lock form: integers, text and bytes, no time.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs", advisory_lock = "email_jobs-{id}")]
struct SendEmail {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    #[field(attempt, generated)]
    attempt: i32,
    #[field(payload)]
    payload: Vec<u8>,
}

/// One insert for every database: the derive's own, on the connection the pool hands out.
impl<DB> Publish<DB> for SendEmail
where
    DB: QueueDatabase,
    Self: Insert<DB::Connection>,
{
    async fn publish(
        conn: &mut DB::Connection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        let job = Self {
            id: 0,
            name: message.name().to_owned(),
            attempt: 1,
            payload: message.payload().to_vec(),
        };
        job.insert(conn).await
    }
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    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::<Any>::from_url(url, AnyPoolOptions::new());
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = AnyConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<SendEmail>("emails"), |b| {
            b.include(send);
        })
}

#[ruststream::app]
fn app() -> impl App {
    // An `AnyPool` reaches its driver by the URL's scheme.
    install_default_drivers();
    let url = env::var("DATABASE_URL").unwrap_or_else(|_| "sqlite:mail.db?mode=rwc".to_owned());
    service(&url)
}

Manual

examples/manual/sqlx_databases_any.rs

use ruststream::OutgoingMessage;
use ruststream_sqlx::dialect::{Column, KeyPart};
use ruststream_sqlx::spec::{Advisory, Attempt, Payload};
use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable, PayloadRow};
use sqlx::any::AnyPoolOptions;
use sqlx::{Any, AnyConnection, Connection, Database, Postgres};

/// One row of `email_jobs` in the advisory lock form: integers, text and bytes, no time.
#[derive(sqlx::FromRow)]
struct SendEmail {
    id: i64,
    attempt: i32,
    payload: Vec<u8>,
}

impl InboxTable for SendEmail {
    type Id = i64;
    type Table = InboxSpec<(Advisory, Attempt, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("email_jobs", Column::new("id").generated())
        .group(Column::new("name"))
        .advisory(&[KeyPart::Literal("email_jobs-"), KeyPart::Column("id")])
        .attempt(Column::new("attempt").generated())
        .payload(Column::new("payload"));

    fn id(&self) -> &i64 {
        &self.id
    }
}

impl AttemptRow for SendEmail {
    type Attempt = i32;

    fn attempt(&self) -> &i32 {
        &self.attempt
    }
}

impl PayloadRow for SendEmail {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

/// `sqlx::Any` hands a statement to its backend as written, so the insert names the
/// placeholders of the database the connection reaches.
impl Publish<Any> for SendEmail {
    async fn publish(
        conn: &mut AnyConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        let insert = if conn.backend_name() == Postgres::NAME {
            "INSERT INTO email_jobs (name, payload) VALUES ($1, $2)"
        } else {
            "INSERT INTO email_jobs (name, payload) VALUES (?, ?)"
        };
        sqlx::query(insert)
            .bind(message.name())
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

struct Mailer;

impl Handle<Email> for Mailer {
    fn handle(
        &self,
        email: &Email,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(to = %email.to, "sending");
        ready(Ok(()))
    }
}

/// 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::<Any>::from_url(url, AnyPoolOptions::new());
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = AnyConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<SendEmail>("emails"), |b| {
            let send =
                subscriber("emails", Mailer).map_source(|_| InboxQueue::<SendEmail>::new("emails"));
            b.include(send.build());
        })
}

fn main() -> ExitCode {
    // An `AnyPool` reaches its driver by the URL's scheme.
    install_default_drivers();
    let url = env::var("DATABASE_URL").unwrap_or_else(|_| "sqlite:mail.db?mode=rwc".to_owned());
    run_main(|| service(&url))
}