Struct Tx 

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

The delivery’s transaction, lent to the handler: it dereferences to the connection the transaction runs on, and goes back to the delivery when the handler ends.

A handler takes it as its first Ctx parameter, Ctx(mut tx): Ctx<keys::Tx<DB>>, on a registration mounted with transactional. It writes through &mut *tx, as through the connection of a sqlx transaction, and (&mut *tx).begin() nests a savepoint of its own. The delivery’s settlement ends the transaction: acknowledgement commits it, every other outcome rolls the handler’s writes back first. Dropping Tx gives the transaction back and ends nothing. A Tx dropped by a panic marks the handler’s writes for the rollback, so a delivery acknowledged after its handler panicked, as the skip panic policy does, commits none of them.

Tx is an owned value, so a handler may move it into a task of its own. The transaction must be back when the delivery settles: a settlement that finds it still lent fails with SqlxBrokerError::TransactionHeld, and the transaction rolls back when that Tx drops.

Examples

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

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "payment_jobs")]
pub struct PaymentJob {
    #[field(id)]
    id: i64,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(serde::Deserialize)]
struct Payment {
    account: String,
    cents: i64,
}

#[subscriber(InboxQueue::<PaymentJob>::new("payments"))]
async fn post(payment: &Payment, Ctx(mut tx): Ctx<keys::Tx<Postgres>>) -> HandlerOutcome {
    let posted = sqlx::query("INSERT INTO ledger (account, cents) VALUES ($1, $2)")
        .bind(&payment.account)
        .bind(payment.cents)
        .execute(&mut *tx)
        .await;
    match posted {
        // The commit keeps the ledger row and finishes the job, together.
        Ok(_) => HandlerOutcome::ack(),
        // The retry discards the ledger row and returns the job to the queue.
        Err(error) => {
            tracing::warn!(%error, "the payment was not posted");
            HandlerOutcome::retry()
        }
    }
}

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

Trait Implementations

impl<DB: QueueDatabase> AsMut<<DB as Database>::Connection> for Tx<DB>

fn as_mut(&mut self) -> &mut DB::Connection

Converts this type into a mutable reference of the (usually inferred) input type.

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

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

Formats the value using the given formatter.

impl<DB: QueueDatabase> Deref for Tx<DB>

fn deref(&self) -> &DB::Connection

The connection inside the delivery’s transaction.

Panics

Never: Tx holds the transaction until it drops.

type Target = <DB as Database>::Connection

The resulting type after dereferencing.

impl<DB: QueueDatabase> DerefMut for Tx<DB>

fn deref_mut(&mut self) -> &mut DB::Connection

The connection inside the delivery’s transaction, to run a statement on.

Panics

Never: Tx holds the transaction until it drops.

impl<DB: QueueDatabase> Drop for Tx<DB>

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.