Struct TxContext 

pub struct TxContext<DB: QueueDatabase> { /* private fields */ }

What a transactional delivery on DB lends its handler: the context Ctx<keys::Tx<DB>> reads, and Ctx<keys::Pool<DB>> and Ctx<keys::Attempt> after it.

It is built only for a handler whose first Ctx reads it, and copies two references, where the delivery’s transaction waits and the pool, with the transaction’s slot and the attempt.

Examples

use ruststream_sqlx::prelude::*;
use sqlx::{PgPool, Postgres};

// `keys::Tx` comes first, so the handler's context is `TxContext<Postgres>`, and the keys
// after it read the same context.
#[subscriber(InboxQueue::<Job>::new("transfers"))]
async fn transfer(
    order: &Transfer,
    Ctx(mut tx): Ctx<keys::Tx<Postgres>>,
    Ctx(pool): Ctx<keys::Pool<Postgres>>,
    Ctx(attempt): Ctx<keys::Attempt>,
) -> HandlerOutcome {
    // Through the pool: the attempt stays logged whatever the outcome.
    let _ = sqlx::query("INSERT INTO transfer_log (sender, attempt) VALUES ($1, $2)")
        .bind(&order.from)
        .bind(attempt.and_then(|attempt| i64::try_from(attempt).ok()))
        .execute(&pool)
        .await;
    // Through the transaction: both legs commit with the acknowledgement, or neither does.
    let debited = sqlx::query("UPDATE accounts SET cents = cents - $1 WHERE name = $2")
        .bind(order.cents)
        .bind(&order.from)
        .execute(&mut *tx)
        .await;
    if debited.is_err() {
        return HandlerOutcome::retry();
    }
    let credited = sqlx::query("UPDATE accounts SET cents = cents + $1 WHERE name = $2")
        .bind(order.cents)
        .bind(&order.to)
        .execute(&mut *tx)
        .await;
    if credited.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("bank", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(transfer.transactional());
    })
}

Trait Implementations

impl<DB, Row, Mode> BuildContext<InboxDelivery<DB, Row, Mode>> for TxContext<DB>

where DB: QueueDatabase, Row: Events<DB>, InboxDelivery<DB, Row, Mode>: TransactionalDelivery<DB>,

fn build(msg: &InboxDelivery<DB, Row, Mode>) -> Self

Builds the context value by reading fields out of msg.

impl<DB: QueueDatabase> Clone for TxContext<DB>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<DB: QueueDatabase> Copy for TxContext<DB>

impl<DB: QueueDatabase> Debug for TxContext<DB>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<S: Sync, DB: QueueDatabase> FromContext<TxContext<DB>, S> for Ctx<Attempt>

type Rejection = !

The error returned when extraction fails. It is converted into a HandlerOutcome (the reflexive conversion makes HandlerOutcome itself a valid rejection, and Infallible works for an extractor that never fails) and the delivery is settled by that outcome, skipping the handler body.

fn from_context( ctx: &mut Context<'_, TxContext<DB>, S>, ) -> impl Future<Output = Result<Self, Infallible>> + Send

Resolves the value from the delivery context. The context is borrowed mutably so an extractor may also read broker fields or take scratch a middleware left for it.

impl<S: Sync, DB: QueueDatabase> FromContext<TxContext<DB>, S> for Ctx<Pool<DB>>

type Rejection = !

The error returned when extraction fails. It is converted into a HandlerOutcome (the reflexive conversion makes HandlerOutcome itself a valid rejection, and Infallible works for an extractor that never fails) and the delivery is settled by that outcome, skipping the handler body.

fn from_context( ctx: &mut Context<'_, TxContext<DB>, S>, ) -> impl Future<Output = Result<Self, Infallible>> + Send

Resolves the value from the delivery context. The context is borrowed mutably so an extractor may also read broker fields or take scratch a middleware left for it.