Struct InboxSubscriber 

pub struct InboxSubscriber<DB: QueueDatabase, Row: Events<DB>, Mode = Plain> { /* private fields */ }

The subscriber an InboxQueue opens: a stream of deliveries claimed from the table.

The stream claims when it is polled: up to the batch size for a batch handler, and one row for a single-message handler, with a claim in flight for each free worker of a handler mounted with workers(n). The subscription holds at most the pool’s size less one connection, and a claim beside another one starts only while the pool has a connection to spare, so the pool keeps one for the handlers. It keeps one claim in flight on SQLite, which takes one writer at a time, and on a table whose groups keep their order or that has a partition_key column, whose order claims that run at once could break, and where a lease claim stamps its rows in a transaction, as on MySQL and MariaDB. After a claim that filled its limit the next one runs at once; after one that found fewer rows it waits the poll interval, or until a publisher of the same broker writes a row of its table and group (a row the service writes through its own SQL or a handler’s transaction waits for the interval). A write that lands while the subscription claims ends its next wait at once. A failed claim reaches the stream as an error item and the next claim waits one second. After shutdown the stream ends.

In the row lock form a message in work holds one connection, and a batch one for all its messages. A batch’s settlements take effect together, when the last of its deliveries finishes: a settlement whose statement fails rolls the whole batch back, and its rows return. A batch that holds a row whose statement always fails therefore rolls back and returns all of its rows on every attempt, until the service’s SQL or schema is fixed.

In the lease form the claim commits at once and holds no connection: each delivery, in a batch or alone, takes a connection only while it settles, and its settlement takes effect on its own; in transactional mode a delivery holds its transaction’s connection until it settles. A task on the runtime the broker connected on extends the lease of every delivery in work each half lease, on one connection, until the subscriber drops or the broker shuts down.

In the advisory lock form each message in work holds a connection of its own, in a batch too, and its session holds the lock on the row’s key: outside transactional mode no transaction stays open while the handler works. A claim selects candidates, locks each key on a session and takes the row while it is still claimable; a batch takes idle connections first and opens new ones while the pool has room, so a batch larger than the pool shrinks instead of waiting. A settlement runs its statement on the session, then releases the lock and returns the connection to the pool. A delivery dropped unsettled closes its connection on the runtime the broker connected on, after releasing the lock, and its row returns at once. SQLite keeps no such locks: the process keeps the keys in work instead, so one process serves a database file.

In transactional mode, a registration mounted with transactional, each delivery lends its handler the transaction it settles in, opened at the table’s isolation level or SQLite mode. In the row lock form that is the claim’s: the subscription sets a savepoint after each claim that took a row, so a settlement other than acknowledgement can discard what the handler wrote. In the lease form the subscription opens a transaction for each delivery once its claim committed; in the advisory lock form, on the delivery’s session once its row is taken. Transactional mode serves single deliveries: the subscriber delivers no batches in it.

A table whose groups keep their order (#[field(group, fifo = true)]) has one row of a group in work at a time. A claim takes the group’s head, its first unfinished row in claim order, and takes nothing while another row of the group is in work, a row that entered the group ahead of it included. The claim first takes the group in its transaction with the guard its dialect builds, so subscriptions to one group on separate brokers keep the order too. A batch of such a table holds one row, and a handler mounted with workers(n) handles the group’s rows one after another.

Examples

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

/// Changes to push to a CRM, one group per customer, each group in order.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "crm_jobs")]
pub struct SyncChange {
    #[field(id, generated)]
    id: i64,
    #[field(group, fifo = true)]
    customer: String,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Change {
    field: String,
}

// The mount's subscription is an `InboxSubscriber<Postgres, SyncChange>` reading group `acme`.
#[subscriber(InboxQueue::<SyncChange>::new("acme"))]
async fn push(change: &Change) -> HandlerOutcome {
    tracing::info!(field = %change.field, "pushing");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("crm-sync", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        // Four workers, and still one change of the group at a time, in order.
        b.include(push.workers(nonzero!(4)));
    })
}

Trait Implementations

impl<DB, Row> BatchSubscriber for InboxSubscriber<DB, Row, Plain>

where DB: QueueDatabase, Row: Events<DB>, Row::Lane: BatchLane<DB, Row>,

type Batch = <<Row as QueueRow>::Lane as BatchLane<DB, Row>>::Batch

Container yielded by batches. Implementations choose between Vec, custom iterators, or anything else that yields the underlying Subscriber::Message.

fn batches( &mut self, size: NonZeroUsize, ) -> impl Stream<Item = Result<Self::Batch, Self::Error>> + Send + '_

Returns a stream of batches of at most size messages.

impl<DB: QueueDatabase, Row: Events<DB>, Mode> Debug for InboxSubscriber<DB, Row, Mode>

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

Formats the value using the given formatter.

impl<DB, Row, Mode> Subscriber for InboxSubscriber<DB, Row, Mode>

where DB: QueueDatabase, Row: Events<DB>, Mode: InboxMode,

type Message = InboxDelivery<DB, Row, Mode>

The message type yielded by this subscriber.

type Error = SqlxBrokerError

The error type yielded by the stream when delivery fails.

fn stream( &mut self, ) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_

Returns a stream of broker deliveries.