Struct InboxQueue 

pub struct InboxQueue<Row, Mode = Plain> { /* private fields */ }

A subscription to a queue table: the rows of Row that the name addresses.

With a group field the name selects the group; without one the table is a single queue and the name is its address. By default rows are claimed with FOR UPDATE SKIP LOCKED in a transaction held for the whole handler: acknowledgement is the delete (or the processed_at mark) and the commit, a retry counts the attempt and commits, and after a crash the database rolls back and the row returns at once. That transaction opens at the table’s isolation where the struct declares one, and a subscription to a table at a level the broker’s dialect does not open (Opens) does not compile.

Mounted with transactional, a subscription lends its handler a transaction to write through, in every form: acknowledgement commits the handler’s writes with the row’s settlement, and every other outcome discards them first. A row lock subscription lends the claim’s transaction.

A table with a #[field(locked_until)] field is claimed by lease instead: the claim writes the lease’s expiry into the row, counts the attempt and commits at once, so the lease, not a transaction, holds the row while the handler runs. While the handler runs, the subscription extends the lease each half lease. Each settlement is one statement that takes effect only while the row still holds the lease, a delivery dropped unsettled releases its row at once, and after a crash the row returns once the lease runs out. The lease is the broker’s (SqlxBroker::lease) unless the subscription sets its own (lease). The broker’s dialect serves a table’s form when it implements the form’s trait: RowLock for the row lock form, Lease for the lease form, Advisory for the advisory lock form. SQLite has no row locks, so a table there takes the lease form or the advisory lock form, and a subscription to one without locked_until or advisory_lock does not compile.

max_attempts(n) and dead_letter(..) at the mount site map onto the table, declared together: at the cap the row moves to the dead_letter group (with a group field) or into the dead_letter table (one with the same columns). With max_attempts(1) every failure moves the row at once. A registration that declares one without the other stops at startup.

A table in payload mode hands its handler the payload, decoded by a codec. A table in row mode, whose struct has no #[field(payload)] field, hands it the row itself: &Row for one delivery, &[Row] for a batch (see Row mode). The forms, transactional mode and the retry declarations work the same in both modes.

Examples

use chrono::{DateTime, Utc};
use ruststream::prelude::*;
use ruststream_sqlx::{Inbox, InboxQueue, SqlxBroker};
use serde::Deserialize;
use sqlx::PgPool;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(retry_after)]
    retry_after: DateTime<Utc>,
    #[field(attempt)]
    attempt: i16,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        // Five attempts, then the row moves to the `emails.failed` group.
        b.include(send).max_attempts(nonzero!(5u32)).dead_letter("emails.failed");
    })
}

In row mode the handler takes the row the subscription claimed:

use ruststream_sqlx::prelude::*;
use sqlx::PgPool;

/// A page to send to the engineer on call: the row itself is the task.
#[derive(Inbox, sqlx::FromRow, Clone)]
#[inbox(table = "page_jobs")]
pub struct Page {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    name: String,
    #[field(attempt, generated)]
    attempt: i16,
    on_call: String,
    text: String,
}

#[subscriber(InboxQueue::<Page>::new("pages"))]
async fn notify(page: &Page) -> HandlerOutcome {
    if send_page(&page.on_call, &page.text).await {
        HandlerOutcome::ack()
    } else {
        HandlerOutcome::retry()
    }
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("pager", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        // Three attempts, then the row moves to the `pages.failed` group.
        b.include(notify).max_attempts(nonzero!(3u32)).dead_letter("pages.failed");
    })
}

Implementations

impl<Row> InboxQueue<Row>

pub fn new(name: impl Into<Cow<'static, str>>) -> Self

A subscription to the queue name of Row’s table, in the plain mode: the handler leaves the delivery’s transaction alone. The mount-site step transactional switches it to transactional mode.

Examples

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Email {
    to: String,
}

// `emails` selects the rows of group `emails`; a table without a group column would deliver all
// its rows here.
#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

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

pub const fn poll_interval(self, interval: Duration) -> Self

How long this subscription waits between claims that found the queue empty; the broker’s interval unless set. After a full claim the next one runs at once. A publish through the broker into the subscription’s table and group ends the wait early (waking a subscription).

Examples

use std::time::Duration;

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

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

#[derive(Deserialize)]
struct Page {
    on_call: String,
}

// An urgent queue looks again every 50 ms when it runs dry; the broker's other queues keep theirs.
#[subscriber(InboxQueue::<Alert>::new("alerts").poll_interval(Duration::from_millis(50)))]
async fn notify(page: &Page) -> HandlerOutcome {
    tracing::warn!(on_call = %page.on_call, "paging");
    HandlerOutcome::ack()
}

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

pub const fn lease(self, lease: Duration) -> Self

where Row: LeaseRow,

How long a claim of this subscription leases each row; the broker’s lease (SqlxBroker::lease) unless set.

A lease is whole seconds, at least one: a shorter one is rounded up. The subscription extends the lease of every delivery in work each half lease, so a handler may outlast it; the lease is how long a row stays out of the queue after its process crashed. A lease runs out under a running handler only when its extensions fail or stop (the database out of reach, the subscription closed, the broker shut down), and the late settlement then fails with SqlxBrokerError::LeaseLost. Only a table with a #[field(locked_until)] field takes a lease; on any other the call does not compile.

Examples

use std::time::Duration;

use chrono::{DateTime, Utc};
use ruststream::prelude::*;
use ruststream_sqlx::{Inbox, InboxQueue, SqlxBroker};
use serde::Deserialize;
use sqlx::PgPool;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "report_jobs")]
pub struct Report {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Request {
    month: u32,
}

// The report of a renderer that crashed waits out a two-minute lease before another
// process takes it.
#[subscriber(InboxQueue::<Report>::new("reports").lease(Duration::from_secs(120)))]
async fn render(request: &Request) -> HandlerOutcome {
    tracing::info!(month = request.month, "rendering");
    HandlerOutcome::ack()
}

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

Trait Implementations

impl<Row, Mode> Clone for InboxQueue<Row, Mode>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<Row, Mode: InboxMode> Debug for InboxQueue<Row, Mode>

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

Formats the value using the given formatter.

impl<DB, D, Row, Mode> SubscriptionSource<ConnectedSqlxBroker<DB, D>> for InboxQueue<Row, Mode>

where DB: QueueDatabase, D: Dialect + Opens<Row::Opening> + 'static, Row: InboxRow + Events<DB>, Row::Form: FormOn<D>, Mode: InboxMode,

type Subscriber = InboxSubscriber<DB, Row, Mode>

The subscriber type this source opens.

type Copies = BrokerMoves

Who publishes the copies this subscription’s retries are made of, and who names where they go: AddressedCopies, NamedCopies or BrokerMoves.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe( self, connected: &ConnectedSqlxBroker<DB, D>, ) -> Result<Self::Subscriber, SqlxBrokerError>

Opens the subscription against the connected broker. Called once at startup.

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

Takes the registration’s retry declaration into the descriptor, before the subscription opens.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.

fn declare_retry_on( &self, _connected: &ConnectedSqlxBroker<DB, D>, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.