Struct NamedDelivery 

pub struct NamedDelivery<DB, D = BuiltIn<DB>>
where
    DB: QueueDatabase + RoleColumns,
    D: ByName<DB> + 'static,
{ /* private fields */ }

A delivery of a by-name subscription: a row of the table its name’s route leads to.

It settles as an InboxDelivery does. A route whose row leaves every event to the crate delivers its rows read by role: no box and no dynamic call per message, as through an InboxQueue. A row that overrides an event, or holds a column type the crate does not read itself, settles through its own code: each delivery is boxed, and each settlement is a dynamic call whose future is boxed too.

Examples

use ruststream::OutgoingMessage;
use ruststream::prelude::*;
use ruststream_sqlx::{Inbox, Publish, SqlxBroker};
use serde::Deserialize;
use sqlx::{PgConnection, PgPool, Postgres};

#[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>,
}

impl Publish<Postgres> for SendEmail {
    async fn publish(
        conn: &mut PgConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO email_jobs (name, payload) VALUES ($1, $2)")
            .bind(message.name())
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

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

// `emails` is a name: its route leads it to `email_jobs`, and each delivery settles there.
#[subscriber("emails")]
async fn send(email: &Email) -> HandlerOutcome {
    if email.to.is_empty() {
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

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

Trait Implementations

impl<DB, D> Debug for NamedDelivery<DB, D>

where DB: QueueDatabase + RoleColumns, D: ByName<DB> + 'static,

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

Formats the value using the given formatter.

impl<DB, D> IncomingMessage for NamedDelivery<DB, D>

where DB: QueueDatabase + RoleColumns, D: ByName<DB> + 'static,

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.