Struct NamedSubscriber 

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

The subscriber a by-name subscription opens: the claim loop of the table its name’s route leads to.

It claims as an InboxSubscriber does, with the broker’s poll interval, on a broker whose dialect implements ByName for its database, as every built-in dialect does. A route whose row leaves every event to the crate, with role columns of types the crate reads itself, is read by those columns alone, in the row lock form and in the lease form: no box and no dynamic call per message. Each column is held to the type the struct reads it as, so a row that would not decode into the struct goes to the decode-failure policy here too. The types are an i16, i32, i64, String or Vec<u8> id, a Vec<u8> or String payload, headers as JSON, a text or byte key, an integer attempt, chrono or time times and leases, and SystemClock or DatabaseClock. Any other row runs its own code: each delivery is boxed, each settlement is a dynamic call whose future is boxed, and each poll of the stream is a dynamic call.

A mount that declares max_attempts(..) or dead_letter(..) on a name is refused at startup: an InboxQueue descriptor takes those.

Examples

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

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

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

#[derive(Deserialize)]
struct Request {
    day: String,
}

// Each by-name mount opens a `NamedSubscriber`; the route's prefix leads both names into
// `report_jobs`, and each reads its own group.
#[subscriber("reports.daily")]
async fn daily(request: &Request) -> HandlerOutcome {
    tracing::info!(day = %request.day, "daily report");
    HandlerOutcome::ack()
}

#[subscriber("reports.weekly")]
async fn weekly(request: &Request) -> HandlerOutcome {
    tracing::info!(day = %request.day, "weekly report");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(
        SqlxBroker::new(pool).route::<Report>("reports.*"),
        |b| {
            b.include(daily);
            b.include(weekly);
        },
    )
}

Trait Implementations

impl<DB, D> Debug for NamedSubscriber<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> Subscriber for NamedSubscriber<DB, D>

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

type Message = NamedDelivery<DB, D>

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.