Trait InboxSettings 

pub trait InboxSettings: Declared {
    // Provided methods
    fn transactional(self) -> <Self::Settings as TransactionalStep>::Out
       where Self::Settings: TransactionalStep { ... }
    fn on_threads(
        self,
        threads: InboxThreads,
    ) -> <<Self::Settings as WorkersStep>::Out as ThreadsStep>::Out
       where Self::Settings: WorkersStep,
             <Self::Settings as WorkersStep>::Out: ThreadsStep { ... }
}

The inbox’s own mount-site steps, on a #[subscriber] definition and on its settings builder.

The prelude brings it in, beside the framework’s own settings.

Examples

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

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

#[derive(serde::Deserialize)]
struct Invoice {
    number: i64,
}

#[subscriber(InboxQueue::<InvoiceJob>::new("invoices"))]
async fn book(invoice: &Invoice, Ctx(mut tx): Ctx<keys::Tx<Postgres>>) -> HandlerOutcome {
    let booked = sqlx::query("INSERT INTO booked (number) VALUES ($1)")
        .bind(invoice.number)
        .execute(&mut *tx)
        .await;
    if booked.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        // The inbox's step chains with the framework's.
        b.include(book.transactional().workers(nonzero!(4)));
    })
}

Provided Methods

fn transactional(self) -> <Self::Settings as TransactionalStep>::Out

where Self::Settings: TransactionalStep,

Switches an InboxQueue subscription to transactional mode: the handler takes the delivery’s transaction (Ctx<keys::Tx<DB>>), and acknowledgement commits what it wrote.

The handler writes through Tx, which dereferences to the connection inside the delivery’s transaction. Acknowledgement commits the handler’s writes together with the settlement of the row. A retry, a delayed retry, a drop and a dead-letter move roll the handler’s writes back first, then settle the row as outside transactional mode, and so does an acknowledgement after the handler panicked. A delivery dropped unsettled ends its transaction, and the row returns at once.

Each form keeps the transaction its own way. The row lock form lends the claim’s transaction: the subscription sets a savepoint right after each claim, one more statement per delivery, so a retry discards the handler’s writes and still counts the attempt. The lease form opens a transaction of the delivery’s own once the claim committed, one more statement per delivery: the acknowledgement runs inside it by the lease the delivery holds, and a delivery whose lease was lost meanwhile rolls everything back and fails with SqlxBrokerError::LeaseLost. The advisory lock form opens the transaction on the session that holds the row’s key, one more statement per delivery, and ends it before the key’s release. Every transaction opens at the table’s isolation level or SQLite mode. A lease table on Postgres at isolation = repeatable_read or serializable refuses transactional mode when it starts: its transaction cannot see the lease extended after its first statement.

Each delivery in work holds one pool connection, so a handler that takes another connection of the same pool needs a pool larger than its workers(n). On SQLite a delivery’s transaction holds the database’s one write lock from its first write, or from its start in immediate mode, until it settles: every other writer waits for it. Transactional mode serves single deliveries: a batch handler mounted with it does not compile.

Examples

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

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

#[derive(serde::Deserialize)]
struct Restock {
    sku: String,
    count: i32,
}

#[subscriber(InboxQueue::<StockJob>::new("restocks"))]
async fn restock(order: &Restock, Ctx(mut tx): Ctx<keys::Tx<Postgres>>) -> HandlerOutcome {
    let raised = sqlx::query("UPDATE stock SET count = count + $1 WHERE sku = $2")
        .bind(order.count)
        .bind(&order.sku)
        .execute(&mut *tx)
        .await;
    // The count rises exactly when the job is finished: both commit together.
    if raised.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
}

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

fn on_threads( self, threads: InboxThreads, ) -> <<Self::Settings as WorkersStep>::Out as ThreadsStep>::Out

where Self::Settings: WorkersStep, <Self::Settings as WorkersStep>::Out: ThreadsStep,

Runs an InboxQueue subscription’s handlers on dedicated threads, as threads(n) does, with a pool of each thread’s own for their queries.

It fills the position workers(n) and threads(n) fill, so the attribute names neither. A handler on a thread reads Ctx<keys::Pool<..>> as the thread’s pool. Its connections open on the thread and close with it, so none outlives the thread in the service’s pool. Claims, settlements, transactions and publishes take the service’s pool. The subscription counts the threads’ connections against the broker’s connection_limit when it opens.

Examples

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

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

#[derive(serde::Deserialize)]
struct Image {
    path: String,
}

#[subscriber(InboxQueue::<ScanJob>::new("scans"))]
async fn scan(image: &Image, Ctx(pool): Ctx<keys::Pool<Postgres>>) -> HandlerOutcome {
    let noted = sqlx::query("INSERT INTO scanned (path) VALUES ($1)")
        .bind(&image.path)
        .execute(&pool)
        .await;
    if noted.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
}

pub fn app(pool: PgPool) -> RustStream {
    // The service opens 30 connections at most: the pool's and the threads' 4 x 2.
    let broker = SqlxBroker::new(pool).connection_limit(nonzero!(30u32));
    RustStream::new(AppInfo::new("scanner", "0.1.0")).with_broker(broker, |b| {
        b.include(scan.on_threads(InboxThreads::new(nonzero!(4)).connections(nonzero!(2))));
    })
}

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl<Def: Declared> InboxSettings for Def