Repository

A publish writes a row through the row's Publish, the service's own INSERT; Repository names the table at compile time, a static call per publish.

Publish<DB> is the statement of a queue table that takes a published message. Its publish(conn, message) receives the message's name, its encoded bytes and its headers, and lays them out in the table's columns. In a table with a group column, the name usually goes there, so a subscription of that group reads the row. The trait has no default: a row without it is not a publisher, and a mount that names it does not compile.

Repository::<Row>::default() is the publish policy of Row's table. A mount site names it wherever a registration publishes: .out_reply(Repository::<Job>::default()) for a reply, .out(Slot, Repository::<Job>::default()) for an Out slot. The runtime pairs it with the connected broker at startup into a RepositoryPublisher. A publisher outside a handler, such as an HTTP endpoint's, comes from SqlxBroker::new(pool).bindable(), bind(Repository::<Job>::default()) and running.publisher(..).

Each publish takes a connection of the pool, runs the row's Publish on it and commits on its own. A handler's transaction does not cover it: a row that must commit with the handler's writes goes in through Insert on the handler's Tx. A message whose header the table cannot hold byte for byte is refused before the statement runs, with SqlxBrokerError::Header; a headers column holds them. A failed statement returns SqlxBrokerError::Publish, which names the message's name, the table and the row type. After shutdown a publisher answers SqlxBrokerError::Closed.

A publish through a repository wakes the subscriptions of its table and group in the same process at once. A name the service does not know at compile time goes through a route instead: the broker's default publisher looks the name up.

Example

An order's confirmation becomes a row of jobs in the group emails, and the second subscription of the service reads it.

Macros

examples/sqlx_publishing_repository.rs

use ruststream::OutgoingMessage;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `jobs`: the group column holds the name a message was published to.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "jobs")]
struct Job {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The service's own insert: the name goes to the group column, the bytes to the payload.
impl Publish<Sqlite> for Job {
    async fn publish(
        conn: &mut SqliteConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO jobs (name, payload) VALUES (?, ?)")
            .bind(message.name())
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[subscriber(InboxQueue::<Job>::new("orders"), reply)]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { order: order.id }
}

#[subscriber(InboxQueue::<Job>::new("emails"))]
async fn email(confirmation: &Confirmation) -> HandlerOutcome {
    tracing::info!(order = confirmation.order, "emailing the confirmation");
    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::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<Job>("orders"), |b| {
            // The confirmation becomes a row of `jobs` in the group `emails`.
            b.include(confirm).out_reply(Repository::<Job>::default());
            b.include(email);
        })
}

Manual

examples/manual/sqlx_publishing_repository.rs

use ruststream::OutgoingMessage;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `jobs`: the group column holds the name a message was published to.
#[derive(sqlx::FromRow)]
struct Job {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Job {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("jobs", Column::new("id").generated())
        .group(Column::new("name"))
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

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

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

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

/// The service's own insert: the name goes to the group column, the bytes to the payload.
impl Publish<Sqlite> for Job {
    async fn publish(
        conn: &mut SqliteConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO jobs (name, payload) VALUES (?, ?)")
            .bind(message.name())
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        ready(Ok(Confirmation { order: order.id }))
    }
}

struct Email;

impl Handle<Confirmation> for Email {
    fn handle(
        &self,
        confirmation: &Confirmation,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order = confirmation.order, "emailing the confirmation");
        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::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<Job>("orders"), |b| {
            let confirm =
                subscriber("orders", Confirm).map_source(|_| InboxQueue::<Job>::new("orders"));
            // The confirmation becomes a row of `jobs` in the group `emails`.
            b.include(confirm.reply().build())
                .out_reply(Repository::<Job>::default());
            let email =
                subscriber("emails", Email).map_source(|_| InboxQueue::<Job>::new("emails"));
            b.include(email.build());
        })
}