Trait OutboxTable

Source

pub trait OutboxTable:
    Sized
    + Send
    + Sync
    + Unpin
    + 'static {
    type Id: Display + FromStr + Send + Sync + 'static;
    type Table: Described;

    const TABLE: Self::Table;

    // Required methods
    fn id(&self) -> &Self::Id;
    fn name(&self) -> &str;
    fn payload(&self) -> &[u8] ⓘ;
}

Expand description

An outbox record described by hand: the trait #[derive(Outbox)] implements, and what a service implements instead of the derive.

TABLE describes the table with OutboxSpec, and type Table names the builder’s final type; the compiler holds the chain to it. The registry builds the default statements from TABLE once, when the record type is registered, and runs the events the type names with OutboxSpec::own from the record’s own impls. The record reads its row through sqlx::FromRow, and holds only the columns it reads.

§Examples

use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::outbox::spec::ProcessedAt;
use ruststream_sqlx::outbox::{self, Outbox, OutboxSpec, OutboxTable, TrackedName};
use sqlx::{PgConnection, PgPool, 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,)>;
    const TABLE: Self::Table = OutboxSpec::new(
        "outbox",
        Column::new("id"),
        Column::new("name"),
        Column::new("payload"),
    )
    .processed_at(Column::new("processed_at"));

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

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

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

impl outbox::Publish<Postgres> for OrderEvent {
    async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
        sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
            .bind(msg.name())
            .bind(msg.payload())
            .fetch_one(conn)
            .await
    }
}

pub struct Orders;

impl TrackedName for Orders {
    const NAME: &'static str = "orders";
}

pub fn app(pool: PgPool) -> impl App {
    let tracking = Outbox::new(pool).track::<OrderEvent, Orders>();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            b.include(fulfil);
            b.after_startup(Publish, tracking.republish());
        })
}