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 every 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.