Pool

keys::Pool gives a handler the service's pool, whose writes commit on their own whatever the outcome; on dedicated threads it is the thread's pool.

Ctx(pool): Ctx<keys::Pool<Sqlite>> binds a Pool<Sqlite>, a handle the read clones; every delivery of an InboxQueue subscription lends it. A write through it commits on its own: a retry, a drop or a rolled-back transaction leaves it in place. That makes it the place for what must outlive the outcome: a delivery log, an error note, a counter. Each query takes a connection of the pool for its run, on top of the ones the subscription holds, so the pool is sized for the handlers' own queries too (Connections). On SQLite a delivery's transaction holds the database's write lock from its first write, so a transactional handler writes through the pool before it writes through tx. On dedicated threads the key yields the thread's own pool, built from the service pool's options, whose connections open on the thread and close with it. After keys::Tx the key still reads the pool, and keys::Attempt may follow it (Context).

Example

Every attempt to send a text message writes a line to the delivery log through the pool, and the line stays when the gateway is busy and the handler retries.

Macros
Manual

examples/sqlx_context_pool.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `sms_jobs`.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "sms_jobs")]
struct SendSms {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<SendSms>::new("sms"))]
async fn send(
    sms: &Sms,
    Ctx(pool): Ctx<keys::Pool<Sqlite>>,
    Ctx(attempt): Ctx<keys::Attempt>,
) -> HandlerOutcome {
    let sent = deliver(sms, attempt).await;
    // The line commits on its own: a retry leaves it in place.
    let _ = sqlx::query("INSERT INTO delivery_log (phone, sent) VALUES (?, ?)")
        .bind(&sms.phone)
        .bind(sent)
        .execute(&pool)
        .await;
    if sent {
        HandlerOutcome::ack()
    } else {
        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());
    RustStream::new(AppInfo::new("sms", "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::<SendSms>("sms"), |b| {
            b.include(send);
        })
}

examples/manual/sqlx_context_pool.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `sms_jobs`.
#[derive(sqlx::FromRow)]
struct SendSms {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

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

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

impl AttemptRow for SendSms {
    type Attempt = i16;

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

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

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

struct SendText;

impl Handle<Sms, (), (), PoolContext<Sqlite>> for SendText {
    async fn handle(
        &self,
        sms: &Sms,
        _outs: &(),
        ctx: &mut Context<'_, PoolContext<Sqlite>>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(pool) = Ctx::<keys::Pool<Sqlite>>::from_context(ctx).await?;
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        let sent = deliver(sms, attempt).await;
        // The line commits on its own: a retry leaves it in place.
        let _ = sqlx::query("INSERT INTO delivery_log (phone, sent) VALUES (?, ?)")
            .bind(&sms.phone)
            .bind(sent)
            .execute(&pool)
            .await;
        if sent {
            Ok(())
        } else {
            Err(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());
    RustStream::new(AppInfo::new("sms", "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::<SendSms>("sms"), |b| {
            let send =
                subscriber("sms", SendText).map_source(|_| InboxQueue::<SendSms>::new("sms"));
            b.include(send.build());
        })
}

Example​

Example