Struct OutboxSpec 

pub struct OutboxSpec<Settings = ()> { /* private fields */ }

The description of an outbox table, with its settings as types.

new names the table and its three required columns: the id, the name a record was published under, and the payload. The column-only setters (within, data, selecting_all) keep the type. Each typed setter (headers, processed_at, own) adds its marker of outbox::spec to the type, and a setting set twice does not compile. The outbox reads the database’s clock. Every setter is a const fn.

Examples

use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::outbox::OutboxSpec;
use ruststream_sqlx::outbox::spec::{Headers, ProcessedAt, own};

/// An outbox in the `shop` schema whose records keep their headers, are marked when
/// processed, and are recovered by the service's own statement.
pub const SHOP_OUTBOX: OutboxSpec<(Headers, ProcessedAt, own::Recover)> = OutboxSpec::new(
    "outbox",
    Column::new("id"),
    Column::new("topic"),
    Column::new("body"),
)
.within("shop")
.headers(Column::new("headers"))
.processed_at(Column::new("processed_at"))
.own::<own::Recover>();

Implementations

impl OutboxSpec

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

The outbox table table, whose records are identified by id, published under the name in name, with the bytes in payload.

Examples

use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::outbox::OutboxSpec;

/// The plainest outbox: a processed record is deleted.
pub const OUTBOX: OutboxSpec = OutboxSpec::new(
    "outbox",
    Column::new("id"),
    Column::new("name"),
    Column::new("payload"),
);

impl<Settings> OutboxSpec<Settings>

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

The table’s description, as the dialects read it.

Examples

use ruststream_sqlx::dialect::{Column, OutboxDialect, Postgres, StatementError};
use ruststream_sqlx::outbox::OutboxSpec;

const OUTBOX: OutboxSpec = OutboxSpec::new(
    "outbox",
    Column::new("id"),
    Column::new("name"),
    Column::new("payload"),
);

/// The statement the outbox's default recovery runs on Postgres.
pub fn recovery() -> Result<String, StatementError> {
    Ok(Postgres.outbox_recover(&OUTBOX.spec())?.sql().to_owned())
}

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

The table in schema.

Examples

use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::outbox::OutboxSpec;

pub const OUTBOX: OutboxSpec = OutboxSpec::new(
    "outbox",
    Column::new("id"),
    Column::new("name"),
    Column::new("payload"),
)
.within("shop");

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

The columns the record reads besides its roles; the default fetch and recovery select them.

Examples

use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::outbox::OutboxSpec;

/// A record that also reads when it was created.
pub const OUTBOX: OutboxSpec = OutboxSpec::new(
    "outbox",
    Column::new("id"),
    Column::new("name"),
    Column::new("payload"),
)
.data(&[Column::new("created_at")]);

pub const fn selecting_all(self) -> Self

The default fetch and recovery select every column (SELECT *): for a record whose columns the description cannot list, such as one that flattens another struct.

Examples

use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::outbox::OutboxSpec;

pub const OUTBOX: OutboxSpec = OutboxSpec::new(
    "outbox",
    Column::new("id"),
    Column::new("name"),
    Column::new("payload"),
)
.selecting_all();

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

where Settings: Push<Headers>, <Settings as Push<Headers>>::Out: Declaration,

The column the published headers are kept in. The record implements HeaderRow, and a republished record carries them again.

Examples

use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::outbox::OutboxSpec;
use ruststream_sqlx::outbox::spec::Headers;

pub const OUTBOX: OutboxSpec<(Headers,)> = OutboxSpec::new(
    "outbox",
    Column::new("id"),
    Column::new("name"),
    Column::new("payload"),
)
.headers(Column::new("headers"));

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

where Settings: Push<ProcessedAt>, <Settings as Push<ProcessedAt>>::Out: Declaration,

The column the default acknowledgement writes the database’s current time to; the default fetch and recovery read only the records where it is NULL. Without it a processed record is deleted.

Examples

use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::outbox::OutboxSpec;
use ruststream_sqlx::outbox::spec::ProcessedAt;

pub const OUTBOX: OutboxSpec<(ProcessedAt,)> = OutboxSpec::new(
    "outbox",
    Column::new("id"),
    Column::new("name"),
    Column::new("payload"),
)
.processed_at(Column::new("processed_at"));

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

where Settings: Push<Event>, <Settings as Push<Event>>::Out: Declaration,

The record runs Event itself, through its impl of the event’s trait of outbox, instead of the crate’s default; a record that names the event without the impl is not registered.

Examples

use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::outbox::spec::{ProcessedAt, own};
use ruststream_sqlx::outbox::{self, OutboxSpec, OutboxTable};
use sqlx::{PgConnection, Postgres};

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

impl OutboxTable for OrderEvent {
    type Id = i64;
    type Table = OutboxSpec<(ProcessedAt, own::Retry)>;
    const TABLE: Self::Table = OutboxSpec::new(
        "outbox",
        Column::new("id"),
        Column::new("name"),
        Column::new("payload"),
    )
    .processed_at(Column::new("processed_at"))
    .own::<own::Retry>();

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

    fn name(&self) -> &str {
        &self.name
    }

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

// The service counts how often a consumer asked for a record again.
impl outbox::Retry<Postgres> for OrderEvent {
    async fn retry(conn: &mut PgConnection, id: &i64) -> sqlx::Result<()> {
        sqlx::query("UPDATE outbox SET retries = retries + 1 WHERE id = $1")
            .bind(id)
            .execute(conn)
            .await?;
        Ok(())
    }
}

Trait Implementations

impl<Settings> Clone for OutboxSpec<Settings>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<Settings> Copy for OutboxSpec<Settings>

impl<Settings> Debug for OutboxSpec<Settings>

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

Formats the value using the given formatter.

impl<Settings: Declaration + 'static> Described for OutboxSpec<Settings>

type Declared = Settings

The settings, folded.

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

The table’s description, as the dialects read it.

impl<Settings> Eq for OutboxSpec<Settings>

impl<Settings> PartialEq for OutboxSpec<Settings>

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

Equality operator ==.

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

Inequality operator !=.