Context

A handler reads the inbox's context through Ctx<keys::..>: the transaction, the pool and the attempt, in that order, leaving out what it does not need.

The keys module holds three keys, each read as a handler parameter:

Key	Value	What it is
`keys::Tx<DB>`	`Tx<DB>`	the delivery's transaction, in transactional mode
`keys::Pool<DB>`	`Pool<DB>`	the service's pool, whose writes commit on their own
`keys::Attempt`	`Option<u64>`	the delivery's attempt, 1 for the first, `None` where the table counts none

A handler's first Ctx names the context the delivery builds for it, and decides which keys compile after it. After keys::Tx come the other two, after keys::Pool only keys::Attempt, and nothing after keys::Attempt. So the order Tx, Pool, Attempt fits every handler, with the keys it does not need left out, and a key in another place does not compile. A key's type names its database, as in keys::Pool<Sqlite>. A delivery builds the context only for a handler that reads it: a pool handle and the attempt are copied, and the transaction is lent. The framework's extractors stand beside these keys: Headers<T>, State<T>, and a &mut Context parameter whose headers() read the delivery's headers. In the manual form the body resolves a key with Ctx::<keys::Attempt>::from_context(ctx), its Handle impl naming the context: keys::InboxContext for the attempt, keys::PoolContext<DB> after the pool, keys::TxContext<DB> after the transaction.

Example

A transactional signup handler logs every attempt through the pool and opens the account through the delivery's transaction; a retry keeps the log and discards the account.

Macros
Manual

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

/// One row of `signup_jobs`.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "signup_jobs")]
struct OpenAccount {
    #[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::<OpenAccount>::new("signups"))]
async fn open_account(
    signup: &Signup,
    Ctx(mut tx): Ctx<keys::Tx<Sqlite>>,
    Ctx(pool): Ctx<keys::Pool<Sqlite>>,
    Ctx(attempt): Ctx<keys::Attempt>,
) -> HandlerOutcome {
    // Through the pool: the line stays whatever the outcome. It comes first, because on SQLite
    // the delivery's transaction holds the database's write lock from its first write.
    let _ = sqlx::query("INSERT INTO signup_log (email, attempt) VALUES (?, ?)")
        .bind(&signup.email)
        .bind(attempt.and_then(|attempt| i64::try_from(attempt).ok()))
        .execute(&pool)
        .await;
    // Through the transaction: the account commits with the acknowledgement, or not at all.
    let opened = sqlx::query("INSERT INTO accounts (email) VALUES (?)")
        .bind(&signup.email)
        .execute(&mut *tx)
        .await;
    if opened.is_err() || !register(signup, attempt).await {
        return HandlerOutcome::retry();
    }
    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("accounts", "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::<OpenAccount>("signups"), |b| {
            b.include(open_account.transactional());
        })
}

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

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

impl InboxTable for OpenAccount {
    type Id = i64;
    type Table = InboxSpec<(Attempt, Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("signup_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 OpenAccount {
    type Attempt = i16;

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

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

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

struct OpenAccountHandler;

impl Handle<Signup, (), (), TxContext<Sqlite>> for OpenAccountHandler {
    async fn handle(
        &self,
        signup: &Signup,
        _outs: &(),
        ctx: &mut Context<'_, TxContext<Sqlite>>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(mut tx) = Ctx::<keys::Tx<Sqlite>>::from_context(ctx).await?;
        let Ctx(pool) = Ctx::<keys::Pool<Sqlite>>::from_context(ctx).await?;
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        // Through the pool: the line stays whatever the outcome. It comes first, because on
        // SQLite the delivery's transaction holds the database's write lock from its first write.
        let _ = sqlx::query("INSERT INTO signup_log (email, attempt) VALUES (?, ?)")
            .bind(&signup.email)
            .bind(attempt.and_then(|attempt| i64::try_from(attempt).ok()))
            .execute(&pool)
            .await;
        // Through the transaction: the account commits with the acknowledgement, or not at all.
        let opened = sqlx::query("INSERT INTO accounts (email) VALUES (?)")
            .bind(&signup.email)
            .execute(&mut *tx)
            .await;
        if opened.is_err() || !register(signup, attempt).await {
            return Err(HandlerOutcome::retry());
        }
        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("accounts", "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::<OpenAccount>("signups"), |b| {
            let open_account = subscriber("signups", OpenAccountHandler)
                .map_source(|_| InboxQueue::<OpenAccount>::new("signups"))
                .transactional();
            b.include(open_account.build());
        })
}

Example​

Example