Trait ByName

Source

pub trait ByName<DB: Database>: Dialect {
    // Required methods
    fn headers(value: DB::ValueRef<'_>) -> Result<HeaderMap, BoxDynError>;
    fn bind_time(
        arguments: &mut DB::Arguments,
        time: NamedTime,
    ) -> Result<(), Error>;
}

Expand description

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.

A by-name subscription (#[subscriber("emails")]) reads the table the name’s route leads to. Where the route’s row leaves every event to the crate, the subscription reads the role columns into a row of the crate’s own and binds its statements from there, with no box and no dynamic call per message. The ids, payloads, keys and attempts it reads are integers, text and bytes, which every sqlx driver decodes alike; the headers kept as JSON and the times are the database’s own types, and this trait reads and binds them. A dialect that implements it for its database gets subscriptions by name; on a dialect that does not, a by-name mount does not compile, and its tables take InboxQueue descriptors.

BuiltIn<DB> implements it for each built-in database. A dialect of the service’s own that wraps a built-in one delegates to it; one for a driver outside sqlx reads and binds through that driver.

§Examples

use ruststream::HeaderMap;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::{BuiltIn, ByName, NamedTime};
use serde::Deserialize;
use sqlx::error::BoxDynError;
use sqlx::postgres::{PgArguments, PgValueRef};
use sqlx::{PgPool, Postgres};

/// A dialect of the service's own over Postgres.
#[derive(Debug)]
pub struct Audited;

// Its rows keep their headers and times as the built-in dialect reads them, so
// `#[subscriber("emails")]` mounts on it.
impl ByName<Postgres> for Audited {
    fn headers(value: PgValueRef<'_>) -> Result<HeaderMap, BoxDynError> {
        <BuiltIn<Postgres> as ByName<Postgres>>::headers(value)
    }

    fn bind_time(arguments: &mut PgArguments, time: NamedTime) -> Result<(), sqlx::Error> {
        <BuiltIn<Postgres> as ByName<Postgres>>::bind_time(arguments, time)
    }
}

#[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("emails")]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

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