Skip to main content

Module outbox

ruststream_sqlx

Module outbox 

Source
Expand description

The transactional outbox: what a service publishes is recorded in its own table until a consumer has processed it, and what was not processed is published again at startup.

A struct deriving Outbox, or implementing OutboxTable by hand, describes the table; outbox!, Outbox::register or Outbox::track registers it under the names it tracks, and the registry hands out the two middlewares and the republish. The events below are what the record’s statements do; a service implements the ones it names as its own (#[outbox(custom(..))], or OutboxSpec::own), and always Publish, which has no default.

Each topic has its page on the site:

§Examples

A service whose replies to orders are tracked over the in-memory broker. Inside #[ruststream::app] the registry starts without a pool, and on_startup hands it one:

use std::convert::Infallible;
use std::io;

use chrono::{DateTime, Utc};
use ruststream::OutgoingMessage;
use ruststream::memory::prelude::*;
use ruststream_sqlx::{Outbox, outbox};
use serde::{Deserialize, Serialize};
use sqlx::postgres::{PgConnection, PgPool, Postgres};

// outbox: id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL, payload BYTEA NOT NULL,
// processed_at TIMESTAMPTZ
#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "outbox")]
struct OrderOutbox {
    #[field(id)]
    id: i64,
    #[field(name)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
}

impl outbox::Publish<Postgres> for OrderOutbox {
    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
    }
}

#[derive(Deserialize)]
struct PlaceOrder {
    id: u64,
}

#[derive(Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct OrderPlaced {
    id: u64,
}

#[subscriber("checkout", reply)]
async fn place(cmd: &PlaceOrder) -> OrderPlaced {
    OrderPlaced { id: cmd.id }
}

#[subscriber("orders")]
async fn fulfil(order: &OrderPlaced) -> HandlerOutcome {
    tracing::info!(order = order.id, "fulfilling the order");
    HandlerOutcome::ack()
}

/// What `#[ruststream::app]` runs.
pub fn app() -> impl App {
    let tracking = outbox! {
        "orders" => OrderOutbox,
    };
    let registry = tracking.clone();

    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .on_startup(async move |()| {
            let pool = PgPool::connect("postgres://localhost/orders")
                .await
                .map_err(io::Error::other)?;
            registry.set_pool(pool.clone()).map_err(io::Error::other)?;
            Ok::<_, io::Error>(pool)
        })
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        .after_shutdown(async move |pool| {
            pool.close().await;
            Ok::<_, Infallible>(())
        })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            b.include(fulfil);
            b.after_startup(Publish, tracking.republish());
        })
}

Re-exports§

pub use events::Ack;
pub use events::Discard;
pub use events::Fetch;
pub use events::Publish;
pub use events::Recover;
pub use events::Retry;
pub use spec::OutboxSpec;
pub use spec::OutboxTable;

Modules§

spec
The settings of an outbox table described by hand, as types.

Structs§

Checked
A name tracked by type, kept in the registry’s type for the compile-time check; it holds no record and forwards every call. Machinery behind Outbox::track.
Nil
The end of the registrations.
Outbox
The outbox of a service: the record type of each name it tracks, and the pool their records live in.
PoolAlreadySet
The outbox’s pool was set already: by Outbox::new or by an earlier set_pool.
Registered
A name the outbox tracks with Record, in front of the registrations Rest.
Republishing
The body of the startup republish, which republish hands to after_startup: it resolves once every unprocessed record went out again, and fails startup with the first record that did not.
TrackedPublisher
A publisher that records what it publishes under a registered name, from wrap.
TrackingLayer
The subscription middleware of an Outbox, from layer.
TrackingPublishLayer
The publish middleware of an Outbox, from publish_layer.

Enums§

OutboxError
A failure of the outbox: a record it could not write or read, or a republish that did not reach the broker.
TrackedPublishError
A publish through wrap that failed: the outbox did not record the message, or the publisher it wraps did not send it.

Constants§

OUTBOX_ID_HEADER
The header a tracked message carries its record’s id in: written with the id’s Display, read back with its FromStr.

Traits§

Ack
Marks the record processed once its handler acknowledged the message.
Discard
Runs when the handler dropped the message.
Fetch
Takes the record id into work when its message reaches a subscription.
OutboxDatabase
A sqlx database the default events of an outbox record run on.
OutboxRow
A record of an outbox table: a struct deriving Outbox or implementing OutboxTable.
Publish
Creates the record of a message a handler publishes under a registered name, and returns its id, which the message then carries in OUTBOX_ID_HEADER.
Recover
Selects the unprocessed records of one name, for the republish at startup.
Retry
Runs when the handler asked for the message again.
TrackedName
A name the outbox tracks, as a type: Outbox::track refuses a name tracked twice while the service compiles.