Prelude

One glob, ruststream_sqlx::prelude, brings the core prelude, the derives, the broker, InboxQueue, its steps, the publish policies and the keys.

use ruststream_sqlx::prelude::*; is the one import a service on the inbox writes. It carries the core's own prelude, so #[subscriber], RustStream, HandlerOutcome and Out come with it. It adds the derives Inbox and InboxHeaders, the broker SqlxBroker, the descriptor InboxQueue, the mount steps of InboxSettings and InboxThreads, the traits Insert and Publish, the publish policies Repository and Routed, and the keys a handler reads.

Publish here is the trait a table implements to receive a publish, not a policy. Another broker's prelude may name its publish policy Publish, so a file that imports both globs names one of them by its path.

The rest is imported by path: the event traits such as Ack and Fetch, DatabaseClock, HeaderColumn, the dialect and spec modules a table described by hand uses, and everything of the outbox.

Example

The descriptor with its poll interval, the attempt key, the derive, the routes and the Routed policy of an Out slot all come from the one glob.

Macros

examples/sqlx_broker_prelude.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

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

#[derive(OutSlot)]
#[publishes(Receipt)]
struct Receipts;

#[subscriber(InboxQueue::<OrderJob>::new("orders").poll_interval(Duration::from_millis(200)))]
async fn confirm(
    order: &Order,
    Out(receipts): Out<impl Publisher, Receipts>,
    Ctx(attempt): Ctx<keys::Attempt>,
) -> HandlerOutcome {
    tracing::info!(order.id, ?attempt, "confirming");
    match receipts
        .message(&Receipt { order: order.id })
        .publish()
        .await
    {
        Ok(()) => HandlerOutcome::ack(),
        Err(error) => {
            tracing::warn!(subscription = "orders", order.id, %error, "no receipt");
            HandlerOutcome::retry()
        }
    }
}

/// 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());
    let broker = broker
        .route::<OrderJob>("orders")
        .route::<ReceiptJob>("receipts");
    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, |b| {
            b.include(confirm).out(Receipts, Routed).build();
        })
}

Manual

examples/manual/sqlx_broker_prelude.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

/// The slot receipts leave through.
struct Receipts;

impl OutSlot for Receipts {
    const NAME: &'static str = "Receipts";
    // `Receipt` fixes its destination, so no transform on this slot may name one.
    type Destination = Reads;
}

impl PublishedThrough<Receipts> for Receipt {}

struct Confirm;

impl<Entry> Handle<Order, (), Outs<(Entry,)>, InboxContext> for Confirm
where
    Entry: OutEntry<Receipts, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        ctx: &mut Context<'_, InboxContext>,
    ) -> Result<(), HandlerOutcome> {
        let Ok(Ctx(attempt)) = Ctx::<keys::Attempt>::from_context(ctx).await;
        tracing::info!(order.id, ?attempt, "confirming");
        let receipt = Receipt { order: order.id };
        outs.get(Receipts)
            .message(&receipt)
            .publish()
            .await
            .map_err(|error| {
                tracing::warn!(subscription = "orders", order.id, %error, "no receipt");
                HandlerOutcome::retry()
            })
    }
}

/// 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());
    let broker = broker
        .route::<OrderJob>("orders")
        .route::<ReceiptJob>("receipts");
    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, |b| {
            let confirm = subscriber("orders", Confirm).map_source(|_| {
                InboxQueue::<OrderJob>::new("orders").poll_interval(Duration::from_millis(200))
            });
            b.include(confirm.build()).out(Receipts, Routed).build();
        })
}