Skip to main content

InboxSpec

ruststream_sqlx

Struct InboxSpec 

Source 
pub struct InboxSpec<Settings = ()> { /* private fields */ }
Expand description

The description of a queue table, with its settings as types.

new starts a table in the row lock form, in row mode. The column-only setters (within, group, priority, data, fetching, selecting_all) keep the type. Each typed setter adds its marker of spec to Settings, so a table’s type Table lists the markers its chain sets, in that order. A setting set twice does not compile. Every setter is a const fn, and spec hands back the dialect’s TableSpec.

§Examples

use chrono::{DateTime, Utc};
use ruststream_sqlx::dialect::{Column, KeyPart, level};
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::{Advisory, Clock, Opens, Payload, RetryAfter};
use ruststream_sqlx::{DatabaseClock, InboxSpec, InboxTable, PayloadRow};
use serde::Deserialize;
use sqlx::{Sqlite, SqlitePool};

#[derive(sqlx::FromRow)]
pub struct SyncJob {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for SyncJob {
    type Id = i64;
    type Table = InboxSpec<(
        Advisory,
        Clock<DatabaseClock>,
        Opens<level::Immediate>,
        RetryAfter<DateTime<Utc>>,
        Payload,
    )>;
    const TABLE: Self::Table = InboxSpec::new("sync_jobs", Column::new("id").generated())
        .advisory(&[KeyPart::Literal("sync-"), KeyPart::Column("tenant")])
        .clock::<DatabaseClock>()
        .opens::<level::Immediate>()
        .retry_after(Column::new("retry_after"))
        .payload(Column::new("payload"));

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

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

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

#[derive(Deserialize)]
struct Sync {
    tenant: String,
}

// One tenant syncs at a time: the lock key holds the tenant.
#[subscriber(InboxQueue::<SyncJob>::new("sync"))]
async fn sync(job: &Sync) -> HandlerOutcome {
    tracing::info!(tenant = %job.tenant, "syncing");
    HandlerOutcome::ack()
}

pub fn app(pool: SqlitePool) -> RustStream {
    RustStream::new(AppInfo::new("sync", "1.0.0")).with_broker(
        SqlxBroker::<Sqlite>::new(pool),
        |b| {
            b.include(sync);
        },
    )
}

Implementations§

Source§

impl InboxSpec

Source

pub const fn new(table: &'static str, id: Column<'static>) -> Self

A table in the connection’s default schema, with the column that identifies a row: in the row lock form, in row mode, with no other setting.

§Examples
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::Payload;
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};

// webhooks: id BIGSERIAL PRIMARY KEY, body BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct Webhook {
    id: i64,
    body: Vec<u8>,
}

impl InboxTable for Webhook {
    type Id = i64;
    type Table = InboxSpec<(Payload,)>;
    // The database numbers the rows, so an insert leaves `id` out.
    const TABLE: Self::Table = InboxSpec::new("webhooks", Column::new("id").generated())
        .payload(Column::new("body"));

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

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

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

impl<Settings> InboxSpec<Settings>

Source

pub const fn spec(&self) -> TableSpec<'static>

The table’s description, which every statement is built from.

§Examples
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::dialect::insert::{self, Sql};
use ruststream_sqlx::spec::Payload;
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};
use sqlx::PgConnection;

#[derive(sqlx::FromRow)]
pub struct ReceiptJob {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for ReceiptJob {
    type Id = i64;
    type Table = InboxSpec<(Payload,)>;
    const TABLE: Self::Table = InboxSpec::new("receipt_jobs", Column::new("id").generated())
        .group(Column::new("name"))
        .payload(Column::new("payload"));

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

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

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

// INSERT INTO "receipt_jobs" ("name", "payload") VALUES ($1, $2)
const INSERT: Sql<128> = insert::postgres(&ReceiptJob::TABLE.spec());

/// Queues a receipt in the order's transaction, so the two commit together.
pub async fn enqueue(tx: &mut PgConnection, receipt: &[u8]) -> Result<(), sqlx::Error> {
    sqlx::query(INSERT.as_str())
        .bind("receipts")
        .bind(receipt)
        .execute(tx)
        .await?;
    Ok(())
}
Source

pub const fn within(self, schema: &'static str) -> Self

The same table, inside schema.

§Examples
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::Payload;
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};

#[derive(sqlx::FromRow)]
pub struct InvoiceJob {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for InvoiceJob {
    type Id = i64;
    type Table = InboxSpec<(Payload,)>;
    // The queue lives beside the tables it serves: `billing.invoice_jobs`.
    const TABLE: Self::Table = InboxSpec::new("invoice_jobs", Column::new("id").generated())
        .within("billing")
        .payload(Column::new("payload"));

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

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

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

pub const fn group(self, column: Column<'static>) -> Self

The same table, split into groups by column; a subscription reads one group.

§Examples
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::Payload;
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};
use serde::Deserialize;

#[derive(sqlx::FromRow)]
pub struct MediaJob {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for MediaJob {
    type Id = i64;
    type Table = InboxSpec<(Payload,)>;
    // One table holds two queues: the `kind` column names the queue of each row.
    const TABLE: Self::Table = InboxSpec::new("media_jobs", Column::new("id").generated())
        .group(Column::new("kind"))
        .payload(Column::new("payload"));

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

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

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

#[derive(Deserialize)]
struct Upload {
    file: String,
}

#[subscriber(InboxQueue::<MediaJob>::new("thumbnails"))]
async fn thumbnail(upload: &Upload) -> HandlerOutcome {
    tracing::info!(file = %upload.file, "drawing a thumbnail");
    HandlerOutcome::ack()
}

#[subscriber(InboxQueue::<MediaJob>::new("transcodes"))]
async fn transcode(upload: &Upload) -> HandlerOutcome {
    tracing::info!(file = %upload.file, "transcoding");
    HandlerOutcome::ack()
}
Source

pub const fn priority(self, column: Column<'static>) -> Self

The same table, with rows claimed in the order of column, a smaller value first.

§Examples
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::Payload;
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};

// tickets: id BIGSERIAL PRIMARY KEY, severity SMALLINT NOT NULL, payload BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct Ticket {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Ticket {
    type Id = i64;
    type Table = InboxSpec<(Payload,)>;
    // An outage (severity 0) is answered before a question (severity 3).
    const TABLE: Self::Table = InboxSpec::new("tickets", Column::new("id").generated())
        .priority(Column::new("severity"))
        .payload(Column::new("payload"));

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

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

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

pub const fn data(self, columns: &'static [Column<'static>]) -> Self

The same table, with the columns of the message’s own data: the columns without a role.

§Examples
use ruststream::runtime::{Input, SoloCarried};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::{InboxSpec, InboxTable};

// shipments: id BIGSERIAL PRIMARY KEY, carrier TEXT NOT NULL, tracking TEXT NOT NULL
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct Shipment {
    id: i64,
    carrier: String,
    tracking: String,
}

impl InboxTable for Shipment {
    type Id = i64;
    type Table = InboxSpec;
    // Row mode: the claim reads these columns, and the handler takes the row.
    const TABLE: Self::Table = InboxSpec::new("shipments", Column::new("id").generated())
        .data(&[Column::new("carrier"), Column::new("tracking")]);

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

impl Input for Shipment {
    type Axis = SoloCarried<Self>;
}

#[subscriber(InboxQueue::<Shipment>::new("shipments"))]
async fn track(shipment: &Shipment) -> HandlerOutcome {
    tracing::info!(carrier = %shipment.carrier, tracking = %shipment.tracking, "tracking");
    HandlerOutcome::ack()
}
Source

pub const fn fetching(self, columns: &'static [Column<'static>]) -> Self

The same table, with the columns a message assembled from header fields reads beside them.

§Examples
use ruststream::HeaderMap;
use ruststream::runtime::{Input, SoloCarried};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec;
use ruststream_sqlx::{HeaderFields, InboxSpec, InboxTable, put_header};

#[derive(Debug, Clone, sqlx::FromRow)]
pub struct RefundHeaders {
    id: i64,
    tenant: String,
    order_id: i64,
}

#[derive(Debug, Clone, sqlx::FromRow)]
pub struct Refund {
    #[sqlx(flatten)]
    headers: RefundHeaders,
    reason: Option<String>,
}

impl InboxTable for Refund {
    type Id = i64;
    type Table = InboxSpec<(spec::HeaderFields,)>;
    // `tenant` and `order_id` become headers; the claim reads `reason` for the message.
    const TABLE: Self::Table = InboxSpec::new("refunds", Column::new("id").generated())
        .data(&[Column::new("tenant"), Column::new("order_id")])
        .fetching(&[Column::new("reason")])
        .header_fields();

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

impl Input for Refund {
    type Axis = SoloCarried<Self>;
}

impl HeaderFields for Refund {
    const NAMES: &'static [&'static str] = &["tenant", "order_id"];

    fn header_map(&self) -> HeaderMap {
        let mut headers = HeaderMap::with_capacity(Self::NAMES.len());
        put_header(&mut headers, "tenant", &self.headers.tenant);
        put_header(&mut headers, "order_id", &self.headers.order_id);
        headers
    }
}
Source

pub const fn selecting_all(self) -> Self

The same table, read with *: the row flattens another struct, so the columns are not all known.

§Examples
use ruststream::runtime::{Input, SoloCarried};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::{InboxSpec, InboxTable};

/// The address type the service shares with its other tables.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct Address {
    street: String,
    city: String,
}

#[derive(Debug, Clone, sqlx::FromRow)]
pub struct Parcel {
    id: i64,
    #[sqlx(flatten)]
    address: Address,
}

impl InboxTable for Parcel {
    type Id = i64;
    type Table = InboxSpec;
    // The claim reads every column, whatever `Address` holds.
    const TABLE: Self::Table =
        InboxSpec::new("parcels", Column::new("id").generated()).selecting_all();

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

impl Input for Parcel {
    type Axis = SoloCarried<Self>;
}
Source

pub const fn lease<Time>( self, column: Column<'static>, ) -> InboxSpec<<Settings as Push<Lease<Time>>>::Out>
where Settings: Push<Lease<Time>>, <Settings as Push<Lease<Time>>>::Out: Declaration,

The same table in the lease form: a claim sets column, the lease’s expiry in Time, and commits. Adds Lease<Time>.

§Examples
use chrono::{DateTime, Utc};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Lease, Payload};
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};

// report_jobs: id BIGSERIAL PRIMARY KEY, locked_until TIMESTAMPTZ NULL, payload BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct ReportJob {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for ReportJob {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, Payload)>;
    // A render that takes minutes holds its row by `locked_until`, not by an open transaction.
    const TABLE: Self::Table = InboxSpec::new("report_jobs", Column::new("id").generated())
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

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

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

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

pub const fn advisory( self, key: &'static [KeyPart<'static>], ) -> InboxSpec<<Settings as Push<Advisory>>::Out>
where Settings: Push<Advisory>, <Settings as Push<Advisory>>::Out: Declaration,

The same table in the advisory lock form: a session lock on the key these parts build from the row holds the row. Adds Advisory.

§Examples
use ruststream_sqlx::dialect::{Column, KeyPart};
use ruststream_sqlx::spec::{Advisory, Payload};
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};

// postings: id BIGSERIAL PRIMARY KEY, account TEXT NOT NULL, payload BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct Posting {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Posting {
    type Id = i64;
    type Table = InboxSpec<(Advisory, Payload)>;
    // One account's postings run one at a time across every process: the key is `ledger-<account>`.
    const TABLE: Self::Table = InboxSpec::new("postings", Column::new("id").generated())
        .advisory(&[KeyPart::Literal("ledger-"), KeyPart::Column("account")])
        .payload(Column::new("payload"));

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

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

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

pub const fn payload( self, column: Column<'static>, ) -> InboxSpec<<Settings as Push<Payload>>::Out>
where Settings: Push<Payload>, <Settings as Push<Payload>>::Out: Declaration,

The same table in payload mode, the message’s bytes read from column. Adds Payload.

§Examples
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::Payload;
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};

// audit_events: id BIGINT AUTO_INCREMENT PRIMARY KEY, body TEXT NOT NULL
#[derive(sqlx::FromRow)]
pub struct AuditEvent {
    id: i64,
    body: String,
}

impl InboxTable for AuditEvent {
    type Id = i64;
    type Table = InboxSpec<(Payload,)>;
    // The message is the JSON text in `body`, which the subscription's codec decodes.
    const TABLE: Self::Table = InboxSpec::new("audit_events", Column::new("id").generated())
        .payload(Column::new("body"));

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

impl PayloadRow for AuditEvent {
    type Column = String;

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

pub const fn partition_key( self, column: Column<'static>, ) -> InboxSpec<<Settings as Push<Key>>::Out>
where Settings: Push<Key>, <Settings as Push<Key>>::Out: Declaration,

The same table, with the delivery’s partition key read from column. Adds Key.

§Examples
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Key, Payload};
use ruststream_sqlx::{InboxSpec, InboxTable, KeyRow, PayloadRow};

// export_jobs: id BIGSERIAL PRIMARY KEY, customer TEXT NOT NULL, payload BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct ExportJob {
    id: i64,
    customer: String,
    payload: Vec<u8>,
}

impl InboxTable for ExportJob {
    type Id = i64;
    type Table = InboxSpec<(Key, Payload)>;
    // Each delivery carries its customer as the partition key, which the dispatch keeps in order.
    const TABLE: Self::Table = InboxSpec::new("export_jobs", Column::new("id").generated())
        .partition_key(Column::new("customer"))
        .payload(Column::new("payload"));

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

impl KeyRow for ExportJob {
    type Key = String;

    fn partition_key(&self) -> &String {
        &self.customer
    }
}

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

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

pub const fn attempt( self, column: Column<'static>, ) -> InboxSpec<<Settings as Push<Attempt>>::Out>
where Settings: Push<Attempt>, <Settings as Push<Attempt>>::Out: Declaration,

The same table, counting a row’s attempts in column. Adds Attempt.

§Examples
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::{Attempt, Payload};
use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable, PayloadRow};
use serde::Deserialize;

// downloads: id BIGSERIAL PRIMARY KEY, tries INT NOT NULL DEFAULT 1, payload BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct Download {
    id: i64,
    tries: i32,
    payload: Vec<u8>,
}

impl InboxTable for Download {
    type Id = i64;
    type Table = InboxSpec<(Attempt, Payload)>;
    // A retry counts in `tries`, and the handler reads the count.
    const TABLE: Self::Table = InboxSpec::new("downloads", Column::new("id").generated())
        .attempt(Column::new("tries").generated())
        .payload(Column::new("payload"));

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

impl AttemptRow for Download {
    type Attempt = i32;

    fn attempt(&self) -> &i32 {
        &self.tries
    }
}

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

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

#[derive(Deserialize)]
struct Image {
    url: String,
}

// An image the fifth attempt cannot fetch is given up.
#[subscriber(InboxQueue::<Download>::new("images"))]
async fn download(image: &Image, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    if fetch(&image.url).await {
        HandlerOutcome::ack()
    } else if attempt.unwrap_or(1) >= 5 {
        HandlerOutcome::drop()
    } else {
        HandlerOutcome::retry()
    }
}
Source

pub const fn headers( self, column: Column<'static>, ) -> InboxSpec<<Settings as Push<Headers>>::Out>
where Settings: Push<Headers>, <Settings as Push<Headers>>::Out: Declaration,

The same table, with the delivery’s headers read from column. Adds Headers.

§Examples
use std::collections::BTreeMap;

use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::{Headers, Payload};
use ruststream_sqlx::{HeaderRow, InboxSpec, InboxTable, PayloadRow};
use serde::Deserialize;
use sqlx::types::Json;

// pushes: id BIGSERIAL PRIMARY KEY, meta JSONB NULL, payload BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct PushJob {
    id: i64,
    meta: Option<Json<BTreeMap<String, String>>>,
    payload: Vec<u8>,
}

impl InboxTable for PushJob {
    type Id = i64;
    type Table = InboxSpec<(Headers, Payload)>;
    // The JSON object in `meta` becomes the delivery's headers.
    const TABLE: Self::Table = InboxSpec::new("pushes", Column::new("id").generated())
        .headers(Column::new("meta"))
        .payload(Column::new("payload"));

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

impl HeaderRow for PushJob {
    type Column = Option<Json<BTreeMap<String, String>>>;

    fn headers_mut(&mut self) -> &mut Self::Column {
        &mut self.meta
    }
}

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

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

#[derive(Deserialize)]
struct Notification {
    title: String,
}

#[subscriber(InboxQueue::<PushJob>::new("pushes"))]
async fn push(note: &Notification, ctx: &mut Context<'_>) -> HandlerOutcome {
    let device = ctx.headers().get_str("device").unwrap_or("all");
    tracing::info!(title = %note.title, device, "pushing");
    HandlerOutcome::ack()
}
Source

pub const fn header_fields( self, ) -> InboxSpec<<Settings as Push<HeaderFields>>::Out>
where Settings: Push<HeaderFields>, <Settings as Push<HeaderFields>>::Out: Declaration,

The same table, with the delivery’s headers built from the row’s header fields, whose columns data and fetching name. Adds HeaderFields.

§Examples
use ruststream::HeaderMap;
use ruststream::runtime::{Input, SoloCarried};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec;
use ruststream_sqlx::{HeaderFields, InboxSpec, InboxTable, put_header};

// sms_jobs: id BIGSERIAL PRIMARY KEY, phone TEXT NOT NULL, locale TEXT NOT NULL, text TEXT NOT NULL
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct Sms {
    id: i64,
    phone: String,
    locale: String,
    text: String,
}

impl InboxTable for Sms {
    type Id = i64;
    type Table = InboxSpec<(spec::HeaderFields,)>;
    // The `locale` column reads as a header, as it would on any other broker.
    const TABLE: Self::Table = InboxSpec::new("sms_jobs", Column::new("id").generated())
        .data(&[Column::new("phone"), Column::new("locale"), Column::new("text")])
        .header_fields();

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

impl Input for Sms {
    type Axis = SoloCarried<Self>;
}

impl HeaderFields for Sms {
    const NAMES: &'static [&'static str] = &["locale"];

    fn header_map(&self) -> HeaderMap {
        let mut headers = HeaderMap::with_capacity(Self::NAMES.len());
        put_header(&mut headers, "locale", &self.locale);
        headers
    }
}

#[subscriber(InboxQueue::<Sms>::new("sms"))]
async fn send(sms: &Sms, ctx: &mut Context<'_>) -> HandlerOutcome {
    let locale = ctx.headers().get_str("locale").unwrap_or("en");
    tracing::info!(phone = %sms.phone, locale, len = sms.text.len(), "sending");
    HandlerOutcome::ack()
}
Source

pub const fn retry_after<Time>( self, column: Column<'static>, ) -> InboxSpec<<Settings as Push<RetryAfter<Time>>>::Out>
where Settings: Push<RetryAfter<Time>>, <Settings as Push<RetryAfter<Time>>>::Out: Declaration,

The same table, with the time a retried row is due in column, in Time. Adds RetryAfter<Time>.

§Examples
use std::time::Duration;

use chrono::{DateTime, Utc};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::{Payload, RetryAfter};
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};
use serde::Deserialize;

// crm_pushes: id BIGSERIAL PRIMARY KEY, due TIMESTAMPTZ NOT NULL DEFAULT now(),
// payload BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct CrmPush {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for CrmPush {
    type Id = i64;
    type Table = InboxSpec<(RetryAfter<DateTime<Utc>>, Payload)>;
    // A row is claimed once `due` has passed, so a delayed retry waits in the table.
    const TABLE: Self::Table = InboxSpec::new("crm_pushes", Column::new("id").generated())
        .retry_after(Column::new("due").generated())
        .payload(Column::new("payload"));

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

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

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

#[derive(Deserialize)]
struct Contact {
    email: String,
}

#[subscriber(InboxQueue::<CrmPush>::new("crm"))]
async fn sync(contact: &Contact) -> HandlerOutcome {
    match upsert(&contact.email).await {
        Ok(()) => HandlerOutcome::ack(),
        // The CRM's rate limit says how long to wait.
        Err(wait) => HandlerOutcome::retry_after(wait),
    }
}
Source

pub const fn processed_at<Time>( self, column: Column<'static>, ) -> InboxSpec<<Settings as Push<ProcessedAt<Time>>>::Out>
where Settings: Push<ProcessedAt<Time>>, <Settings as Push<ProcessedAt<Time>>>::Out: Declaration,

The same table, with the time a row was processed written into column, in Time, so a processed row stays in the table. Adds ProcessedAt<Time>.

§Examples
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Payload, ProcessedAt};
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};
use time::OffsetDateTime;

// payments: id BIGSERIAL PRIMARY KEY, processed_at TIMESTAMPTZ NULL, payload BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct PaymentJob {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for PaymentJob {
    type Id = i64;
    type Table = InboxSpec<(ProcessedAt<OffsetDateTime>, Payload)>;
    // A settled payment stays in the table for the auditors, stamped in `processed_at`.
    const TABLE: Self::Table = InboxSpec::new("payments", Column::new("id").generated())
        .processed_at(Column::new("processed_at").generated())
        .payload(Column::new("payload"));

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

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

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

pub const fn fifo_group( self, column: Column<'static>, ) -> InboxSpec<<Settings as Push<Fifo>>::Out>
where Settings: Push<Fifo>, <Settings as Push<Fifo>>::Out: Declaration,

The same table, split into groups by column, with each group in order: at most one row of a group is in work, taken in claim order. Adds Fifo.

§Examples
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::{Fifo, Payload};
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};
use serde::Deserialize;

// stock_updates: id BIGSERIAL PRIMARY KEY, warehouse TEXT NOT NULL, payload BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct StockUpdate {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for StockUpdate {
    type Id = i64;
    type Table = InboxSpec<(Fifo, Payload)>;
    // Each warehouse is a group, and its updates arrive one at a time, in the order written.
    const TABLE: Self::Table = InboxSpec::new("stock_updates", Column::new("id").generated())
        .fifo_group(Column::new("warehouse"))
        .payload(Column::new("payload"));

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

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

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

#[derive(Deserialize)]
struct Level {
    sku: String,
    count: i64,
}

#[subscriber(InboxQueue::<StockUpdate>::new("berlin"))]
async fn berlin(level: &Level) -> HandlerOutcome {
    tracing::info!(sku = %level.sku, level.count, "berlin stock");
    HandlerOutcome::ack()
}
Source

pub const fn clock<Source: TimeSource>( self, ) -> InboxSpec<<Settings as Push<Clock<Source>>>::Out>
where Settings: Push<Clock<Source>>, <Settings as Push<Clock<Source>>>::Out: Declaration,

The same table, reading now from Source: the statements read the database’s own clock where Source is DatabaseClock. Adds Clock<Source>.

§Examples
use chrono::{DateTime, Utc};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Clock, Payload, RetryAfter};
use ruststream_sqlx::{DatabaseClock, InboxSpec, InboxTable, PayloadRow};

// reminders: id BIGSERIAL PRIMARY KEY, retry_after TIMESTAMPTZ NOT NULL DEFAULT now(),
// payload BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct Reminder {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Reminder {
    type Id = i64;
    type Table = InboxSpec<(Clock<DatabaseClock>, RetryAfter<DateTime<Utc>>, Payload)>;
    // Due times count from the database's `now()`, so hosts whose clocks drift still agree.
    const TABLE: Self::Table = InboxSpec::new("reminders", Column::new("id").generated())
        .clock::<DatabaseClock>()
        .retry_after(Column::new("retry_after").generated())
        .payload(Column::new("payload"));

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

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

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

pub const fn opens<Level: OpeningLevel>( self, ) -> InboxSpec<<Settings as Push<Opens<Level>>>::Out>
where Settings: Push<Opens<Level>>, <Settings as Push<Opens<Level>>>::Out: Declaration,

The same table, with its transactions opened at Level, an isolation level or a SQLite mode. Adds Opens<Level>.

§Examples
use ruststream_sqlx::dialect::{Column, level};
use ruststream_sqlx::spec::{Opens, Payload};
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};

#[derive(sqlx::FromRow)]
pub struct PayoutJob {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for PayoutJob {
    type Id = i64;
    type Table = InboxSpec<(Opens<level::ReadCommitted>, Payload)>;
    // The claim's transaction runs at READ COMMITTED on a database whose default is SERIALIZABLE.
    const TABLE: Self::Table = InboxSpec::new("payout_jobs", Column::new("id").generated())
        .opens::<level::ReadCommitted>()
        .payload(Column::new("payload"));

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

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

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

pub const fn own<Event: OwnEvent>( self, ) -> InboxSpec<<Settings as Push<Event>>::Out>
where Settings: Push<Event>, <Settings as Push<Event>>::Out: Declaration,

The same table, with Event written by the service itself: the row implements the event’s trait. Adds Event, a marker of own.

§Examples
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Payload, own};
use ruststream_sqlx::{Discard, InboxSpec, InboxTable, PayloadRow};
use sqlx::{PgConnection, Postgres};

// signups: id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL, payload BYTEA NOT NULL
#[derive(sqlx::FromRow)]
pub struct Signup {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Signup {
    type Id = i64;
    type Table = InboxSpec<(Payload, own::Discard)>;
    const TABLE: Self::Table = InboxSpec::new("signups", Column::new("id").generated())
        .group(Column::new("name"))
        .payload(Column::new("payload"))
        .own::<own::Discard>();

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

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

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

// A dropped signup moves to the `rejected` group, where support reviews it.
impl Discard<Postgres> for Signup {
    async fn discard(conn: &mut PgConnection, id: &i64) -> Result<(), sqlx::Error> {
        sqlx::query("UPDATE signups SET name = 'rejected' WHERE id = $1")
            .bind(id)
            .execute(conn)
            .await?;
        Ok(())
    }
}

Trait Implementations§

Source§

impl<Settings> Clone for InboxSpec<Settings>

Source§

fn clone(&self) -> Self

Returns a duplicate of the value. Read more
1.0.0 (const: unstable) · Source§

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source. Read more
Source§

impl<Settings> Copy for InboxSpec<Settings>

Source§

impl<Settings> Debug for InboxSpec<Settings>

Source§

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

Formats the value using the given formatter. Read more
Source§

impl<Settings> Eq for InboxSpec<Settings>

Source§

impl<Settings> PartialEq for InboxSpec<Settings>

Source§

fn eq(&self, other: &Self) -> bool

Equality operator ==. Read more
1.0.0 (const: unstable) · Source§

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=. Read more

Auto Trait Implementations§

§

impl<Settings> Freeze for InboxSpec<Settings>
where PhantomData<fn() -> Settings>: Freeze,

§

impl<Settings> RefUnwindSafe for InboxSpec<Settings>
where PhantomData<fn() -> Settings>: RefUnwindSafe,

§

impl<Settings> Send for InboxSpec<Settings>
where PhantomData<fn() -> Settings>: Send,

§

impl<Settings> Sync for InboxSpec<Settings>
where PhantomData<fn() -> Settings>: Sync,

§

impl<Settings> Unpin for InboxSpec<Settings>
where PhantomData<fn() -> Settings>: Unpin,

§

impl<Settings> UnsafeUnpin for InboxSpec<Settings>
where PhantomData<fn() -> Settings>: UnsafeUnpin,

§

impl<Settings> UnwindSafe for InboxSpec<Settings>
where PhantomData<fn() -> Settings>: UnwindSafe,

Blanket Implementations§

Source§

impl<T> Any for T
where T: 'static + ?Sized,

Source§

fn type_id(&self) -> TypeId

Gets the TypeId of self. Read more
Source§

impl<T> Borrow<T> for T
where T: ?Sized,

Source§

fn borrow(&self) -> &T

Immutably borrows from an owned value. Read more
Source§

impl<T> BorrowMut<T> for T
where T: ?Sized,

Source§

fn borrow_mut(&mut self) -> &mut T

Mutably borrows from an owned value. Read more
Source§

impl<T> CloneToUninit for T
where T: Clone,

Source§

unsafe fn clone_to_uninit(&self, dest: *mut u8)

🔬This is a nightly-only experimental API. (clone_to_uninit)
Performs copy-assignment from self to dest. Read more
Source§

impl<T> DynClone for T
where T: Clone,

Source§

fn __clone_box(&self, _: Private) -> *mut ()

Source§

impl<Q, K> Equivalent<K> for Q
where Q: Eq + ?Sized, K: Borrow<Q> + ?Sized,

Source§

fn equivalent(&self, key: &K) -> bool

Checks if this value is equivalent to the given key. Read more
Source§

impl<Q, K> Equivalent<K> for Q
where Q: Eq + ?Sized, K: Borrow<Q> + ?Sized,

Source§

fn equivalent(&self, key: &K) -> bool

Compare self to key and return true if they are equal.
Source§

impl<T> From<T> for T

Source§

fn from(t: T) -> T

Returns the argument unchanged.

Source§

impl<T, W> HasTypeWitness<W> for T
where W: MakeTypeWitness<Arg = T>, T: ?Sized,

Source§

const WITNESS: W = W::MAKE

A constant of the type witness
Source§

impl<T> Identity for T
where T: ?Sized,

Source§

const TYPE_EQ: TypeEq<T, <T as Identity>::Type> = TypeEq::NEW

Proof that Self is the same type as Self::Type, provides methods for casting between Self and Self::Type.
Source§

type Type = T

The same type as Self, used to emulate type equality bounds (T == U) with associated type equality constraints (T: Identity<Type = U>).
Source§

impl<T> Instrument for T

Source§

fn instrument(self, span: Span) -> Instrumented<Self> ⓘ

Instruments this type with the provided Span, returning an Instrumented wrapper. Read more
Source§

fn in_current_span(self) -> Instrumented<Self> ⓘ

Instruments this type with the current Span, returning an Instrumented wrapper. Read more
Source§

impl<T, U> Into<U> for T
where U: From<T>,

Source§

fn into(self) -> U

Calls U::from(self).

That is, this conversion is whatever the implementation of From<T> for U chooses to do.

Source§

impl<T> IntoEither for T

Source§

fn into_either(self, into_left: bool) -> Either<Self, Self> ⓘ

Converts self into a Left variant of Either<Self, Self> if into_left is true. Converts self into a Right variant of Either<Self, Self> otherwise. Read more
Source§

fn into_either_with<F>(self, into_left: F) -> Either<Self, Self> ⓘ
where F: FnOnce(&Self) -> bool,

Converts self into a Left variant of Either<Self, Self> if into_left(&self) returns true. Converts self into a Right variant of Either<Self, Self> otherwise. Read more
Source§

impl<T> PublishedThrough<DefaultSlot> for T

Source§

impl<T> Same for T

Source§

type Output = T

Should always be Self
Source§

impl<T> ToOwned for T
where T: Clone,

Source§

type Owned = T

The resulting type after obtaining ownership.
Source§

fn to_owned(&self) -> T

Creates owned data from borrowed data, usually by cloning. Read more
Source§

fn clone_into(&self, target: &mut T)

Uses borrowed data to replace owned data, usually by cloning. Read more
Source§

impl<T, U> TryFrom<U> for T
where U: Into<T>,

Source§

type Error = !

The type returned in the event of a conversion error.
Source§

fn try_from(value: U) -> Result<T, !>

Performs the conversion.
Source§

impl<T, U> TryInto<U> for T
where U: TryFrom<T>,

Source§

type Error = <U as TryFrom<T>>::Error

The type returned in the event of a conversion error.
Source§

fn try_into(self) -> Result<U, <U as TryFrom<T>>::Error>

Performs the conversion.
Source§

impl<T> WithSubscriber for T

Source§

fn with_subscriber<S>(self, subscriber: S) -> WithDispatch<Self> ⓘ
where S: Into<Dispatch>,

Attaches the provided Subscriber to this type, returning a WithDispatch wrapper. Read more
Source§

fn with_current_subscriber(self) -> WithDispatch<Self> ⓘ

Attaches the current default Subscriber to this type, returning a WithDispatch wrapper. Read more