Struct InboxDelivery 

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

One claimed row in a handler’s hands.

In payload mode the payload is lent from the row, without a copy. In row mode the delivery lends its handler the row itself, as the driver read it, with no codec in between. A headers column becomes the delivery’s headers, and the row it lends holds that column empty. A claimed id without a row, and a row the driver could not read, lend no row: the runtime settles them by the decode policy.

In the row lock form the delivery holds the claim’s transaction: settling it runs one statement and commits, and dropping it unsettled rolls the transaction back, which returns the row to the queue at once. In the lease form the delivery holds the lease its claim wrote, which its subscription extends each half lease: settling it runs one statement on a connection of its own, which takes effect only while the row still holds that lease. A lease delivery dropped unsettled releases its row at once, on the runtime the broker connected on; with that runtime gone, the row returns once the lease runs out.

In the advisory lock form the delivery holds a connection whose session holds the lock on the row’s key: settling it runs one statement on that connection, 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; with that runtime gone the connection closes as it drops, and the server ends its session and its lock. A settlement after shutdown released its lock fails with SqlxBrokerError::Closed.

In transactional mode the delivery lends its handler the transaction it settles in, and the handler’s writes go with the settlement: acknowledgement runs its statement and commits them together; every other settlement first discards what the handler wrote, then settles as outside transactional mode. A statement that fails rolls the whole transaction back, and the row returns at once. In the lease form the acknowledgement runs in the delivery’s own transaction by the lease it holds: a lease lost meanwhile rolls everything back and the settlement fails with SqlxBrokerError::LeaseLost. In the advisory lock form the transaction runs on the session that holds the key, and ends before the key’s release. A delivery dropped unsettled closes its transaction’s connection, and the server rolls the transaction back.

Examples

use std::time::Duration;

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(payload)]
    payload: Vec<u8>,
}

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

// Each claimed row reaches `send` as an `InboxDelivery<Postgres, SendEmail>`, and what the
// handler returns settles it.
#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    match deliver(email).await {
        Sent::Yes => HandlerOutcome::ack(),
        // The mail server asked to come back later: the row returns once the minute passed.
        Sent::Later => HandlerOutcome::retry_after(Duration::from_secs(60)),
        // The address does not exist: the row is finished without a send.
        Sent::Never => HandlerOutcome::drop(),
    }
}

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

In row mode the handler borrows the row the delivery holds, beside what it reads off the delivery:

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

/// A mail to send: the table has no payload column, so a handler takes the row itself.
#[derive(Inbox, sqlx::FromRow, Clone)]
#[inbox(table = "mail_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    to: String,
}

// The delivery lends `send` its row and reports the attempt the table counted.
#[subscriber(InboxQueue::<SendEmail>::new("mail"))]
async fn send(email: &SendEmail, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    if deliver(&email.to).await {
        HandlerOutcome::ack()
    } else if attempt < Some(3) {
        HandlerOutcome::retry()
    } else {
        HandlerOutcome::drop()
    }
}

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

Trait Implementations

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

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

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

Builds the context value by reading fields out of msg.

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, Row, Mode> Carries<Row> for InboxDelivery<DB, Row, Mode>

where DB: QueueDatabase, Row: Events<DB> + QueueRow<Lane = RowLane>, Mode: InboxMode,

fn carried(&self) -> Option<&Row>

The value this delivery holds, or None where it has none.

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

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

Formats the value using the given formatter.

impl<DB: QueueDatabase, Row: Events<DB>, Mode> Drop for InboxDelivery<DB, Row, Mode>

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.

impl<DB, Row, Mode> IncomingMessage for InboxDelivery<DB, Row, Mode>

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

fn payload(&self) -> &[u8]

Returns the raw payload of the message.

fn headers(&self) -> &HeaderMap

Returns the headers attached to the message.

fn decode_error(&self) -> Option<&CodecError>

The error the broker met while reading this delivery, or None when it read the delivery whole.

fn partition_key(&self) -> Option<&[u8]>

Returns the routing key the broker partitioned this message by, or None when the message carries no key.

fn redelivery_count(&self) -> Option<u64>

How many times the broker has delivered this message, counting this delivery. None where the transport counts nothing.

async fn ack(self) -> Result<(), AckError>

Acknowledges successful processing. Consumes the message handle.

async fn nack(self, requeue: bool) -> Result<(), AckError>

Negatively acknowledges the message. When requeue is true the broker should redeliver according to its own retry policy; when false it should drop or dead-letter the message.

fn supports_nack_after(&self) -> bool

Reports whether this transport can honor nack_after natively.

async fn nack_after(self, delay: Duration) -> Result<(), AckError>

Negatively acknowledges the message, asking the broker to redeliver it no sooner than delay from now.