InboxQueue

InboxQueue reads one queue of a table: its name selects a group where the table has a group column and addresses the whole table where it has none.

InboxQueue::<Row>::new(name) takes the struct that describes the table and the queue's name. The descriptor goes where a name goes, as the first argument of #[subscriber(..)]; the manual form mounts it on subscriber(name, body) with map_source. Two settings belong to the descriptor, and both default to the broker's: the poll interval and, on a lease table, the lease. The mount site adds the rest: workers, batches, transactional mode, dedicated threads and a retry cap. A subscription claims rows when the runtime polls its stream, so it takes rows as fast as its handlers settle them. Every claimed row is delivered at least once: the row of a process that crashed returns to the queue by the rules of its table's form. A broker serves one subscription per queue, and one per table without a group column; a second one stops the service at startup. A retry never copies a row: the claim's own statements move it, so the mount site names no retry publisher. A plain name in #[subscriber("emails")] reads the table its route leads to instead, as Names shows.

Example

Two subscriptions read the two groups of one table, and a notice published to sms reaches the handler of sms alone.

Macros
Manual

examples/sqlx_subscriptions_queue.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `notification_jobs`: `channel` is the group, and a subscription's name selects one.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "notification_jobs")]
struct Notify {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    channel: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// Reads the rows of the `email` group.
#[subscriber(InboxQueue::<Notify>::new("email"))]
async fn send_email(notice: &Notice) -> HandlerOutcome {
    tracing::info!(to = %notice.to, "sending an email");
    HandlerOutcome::ack()
}

/// Reads the rows of the `sms` group of the same table.
#[subscriber(InboxQueue::<Notify>::new("sms"))]
async fn send_sms(notice: &Notice) -> HandlerOutcome {
    tracing::info!(to = %notice.to, "sending a text message");
    HandlerOutcome::ack()
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    let broker = broker.route::<Notify>("email").route::<Notify>("sms");
    RustStream::new(AppInfo::new("notifier", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            b.include(send_email);
            b.include(send_sms);
        })
}

examples/manual/sqlx_subscriptions_queue.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `notification_jobs`: `channel` is the group, and a subscription's name selects one.
#[derive(sqlx::FromRow)]
struct Notify {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Notify {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("notification_jobs", Column::new("id").generated())
        .group(Column::new("channel"))
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

    fn id(&self) -> &i64 {
        &self.id
    }
}

impl PayloadRow for Notify {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

/// Reads the rows of the `email` group.
struct SendEmail;

impl Handle<Notice> for SendEmail {
    fn handle(
        &self,
        notice: &Notice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(to = %notice.to, "sending an email");
        ready(Ok(()))
    }
}

/// Reads the rows of the `sms` group of the same table.
struct SendSms;

impl Handle<Notice> for SendSms {
    fn handle(
        &self,
        notice: &Notice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(to = %notice.to, "sending a text message");
        ready(Ok(()))
    }
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    let broker = broker.route::<Notify>("email").route::<Notify>("sms");
    RustStream::new(AppInfo::new("notifier", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            let email =
                subscriber("email", SendEmail).map_source(|_| InboxQueue::<Notify>::new("email"));
            b.include(email.build());
            let sms = subscriber("sms", SendSms).map_source(|_| InboxQueue::<Notify>::new("sms"));
            b.include(sms.build());
        })
}

Example​

Example