Struct Outbox 

pub struct Outbox<DB: Database, Records = Nil> { /* private fields */ }

The outbox of a service: the record type of each name it tracks, and the pool their records live in.

It hands out the subscription middleware (layer), the publish middleware (publish_layer), the startup republish (republish) and wrap for a publisher outside the handlers. Every handle shares one pool: a pool set later, with set_pool, reaches the handles given out before.

A message whose name is not registered passes every handle after a comparison with each registered name, and nothing else. A tracked message reads the pool with one atomic load, takes a connection for its statements, and allocates the id header’s value.

Examples

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

/// What `#[ruststream::app]` runs: the pool is built in `on_startup`, inside the runtime.
pub fn app() -> impl App {
    let tracking = Outbox::<Postgres>::deferred().register::<OrderOutbox>("orders");
    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());
        })
}

Implementations

impl<DB: Database> Outbox<DB, Nil>

pub fn new(pool: Pool<DB>) -> Self

An outbox with no names yet, whose records live in pool.

Examples

/// An app built inside a running Tokio runtime, where the pool exists already.
pub async fn run(pool: PgPool) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let tracking = Outbox::new(pool).register::<OrderOutbox>("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());
        })
        .run()
        .await?;
    Ok(())
}

pub fn deferred() -> Self

An outbox with no names and no pool yet: the service gives it the pool with set_pool, usually in on_startup, where the pool is built.

Examples

use std::io;

pub fn app() -> impl App {
    let tracking = Outbox::<Postgres>::deferred().register::<OrderOutbox>("orders");
    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)
        })
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
        })
}

impl<DB: Database, Records: RecordNames> Outbox<DB, Records>

pub fn register<Record: OutboxTable>( self, name: &'static str, ) -> Outbox<DB, Registered<Record, Records>>

where DB: OutboxDatabase, Registered<Record, Records>: RecordList<DB>,

Tracks the messages published under name with the record type Record, for a name known only at run time; track refuses a repeated name while the service compiles. The record’s default statements are built here, once.

Panics

Panics when name is registered already (each name has one record type), and when Record’s table does not fit a dialect of DB (an identifier longer than the database takes).

Examples

/// One record type tracks both names; each name republishes its own records.
pub fn app(pool: PgPool) -> impl App {
    let tracking = Outbox::new(pool)
        .register::<OrderOutbox>("orders")
        .register::<OrderOutbox>("refunds");
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            b.include(cancel).out_reply(Publish);
            b.after_startup(Publish, tracking.republish());
        })
}

pub fn track<Record: OutboxTable, Name: TrackedName>( self, ) -> Outbox<DB, Registered<Record, Checked<Name, Records>>>

where DB: OutboxDatabase, Records: Lacks<Name>, Registered<Record, Checked<Name, Records>>: RecordList<DB>,

Tracks the messages published under Name::NAME with the record type Record. A name tracked twice does not compile. The record’s default statements are built here, once.

Panics

Panics when the name is registered already through register, whose names the compiler does not see, and when Record’s table does not fit a dialect of DB (an identifier longer than the database takes).

Examples

use ruststream_sqlx::outbox::TrackedName;

pub struct Orders;

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

pub struct Refunds;

impl TrackedName for Refunds {
    const NAME: &'static str = "refunds";
}

/// One record type tracks both names; a third `track::<OrderOutbox, Orders>()` would not
/// compile.
pub fn app(pool: PgPool) -> impl App {
    let tracking = Outbox::new(pool)
        .track::<OrderOutbox, Orders>()
        .track::<OrderOutbox, Refunds>();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            b.include(cancel).out_reply(Publish);
            b.after_startup(Publish, tracking.republish());
        })
}

A name tracked twice stops the build:

use ruststream_sqlx::outbox::TrackedName;

pub struct Orders;

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

pub struct OrdersAgain;

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

fn main() {
    let _ = Outbox::<Sqlite>::deferred()
        .track::<OrderOutbox, Orders>()
        .track::<OrderOutbox, OrdersAgain>();
}

pub fn set_pool(&self, pool: Pool<DB>) -> Result<(), PoolAlreadySet>

Gives the outbox its pool, when it was made with deferred.

Errors

PoolAlreadySet when the outbox has a pool already; the one it has stays.

Examples

use std::io;

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)?;
            // Every handle the registry gave out reads this pool from now on.
            registry.set_pool(pool.clone()).map_err(io::Error::other)?;
            Ok::<_, io::Error>(pool)
        })
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
        })
}

impl<DB: Database, Records: RecordList<DB>> Outbox<DB, Records>

pub fn layer(&self) -> TrackingLayer<DB, Records>

The subscription middleware: a delivery under a registered name that carries an id takes its record before the handler runs, and settles it by the handler’s outcome.

Mounted with RustStream::layer.

Examples

/// The consumer side: `fulfil` runs once per record, and its acknowledgement marks the record.
pub fn app(pool: PgPool) -> impl App {
    let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .layer(tracking.layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(fulfil);
        })
}

pub fn publish_layer(&self) -> TrackingPublishLayer<DB, Records>

The publish middleware: a message a handler publishes under a registered name is recorded before it is sent, and carries its record’s id.

Mounted with RustStream::publish_layer.

Examples

/// The producer side: the reply of `place` is recorded, then sent with its record's id.
pub fn app(pool: PgPool) -> impl App {
    let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
        })
}

pub fn republish<Live: Publisher + 'static>( &self, ) -> impl FnOnce(Live) -> Republishing + Send + 'static

The startup republish of every registered name: an after_startup hook that publishes each unprocessed record again through the scope’s publisher, with its id.

The hook allocates its body once, at startup.

Examples

pub fn app(pool: PgPool) -> impl App {
    let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            // Runs once the subscriptions are open; a record it cannot send fails startup.
            b.after_startup(Publish, tracking.republish());
        })
}

pub fn republish_names<Live: Publisher + 'static>( &self, names: impl IntoIterator<Item = &'static str>, ) -> impl FnOnce(Live) -> Republishing + Send + 'static

The startup republish of the registered names names alone; see republish.

Panics

Panics when a name of names is not registered.

Examples

/// Each broker republishes the names it carries.
pub fn app(pool: PgPool) -> impl App {
    let tracking = outbox! { pool: pool, "orders" => OrderOutbox, "audits" => OrderOutbox };
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            b.after_startup(Publish, tracking.republish_names(["orders"]));
        })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(audit).out_reply(Publish);
            b.after_startup(Publish, tracking.republish_names(["audits"]));
        })
}

pub fn wrap<Live: Publisher>( &self, publisher: Live, ) -> TrackedPublisher<Live, DB, Records>

publisher with the publish middleware’s tracking, for publishes outside the handlers (an HTTP endpoint, an after_startup hook), which the publish pipeline does not reach.

Examples

pub async fn serve(pool: PgPool) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
    let broker = MemoryBroker::new().bindable();
    let egress = broker.bind(Publish);
    let running = RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(broker, |b| {
            b.include(fulfil);
        })
        .start()
        .await?;

    // What an HTTP endpoint publishes: recorded first, then sent with its record's id.
    let publisher = tracking.wrap(running.publisher(egress).await?);
    publisher.message(&OrderPlaced { id: 7 }).publish().await?;

    running.shutdown().await?;
    Ok(())
}

Trait Implementations

impl<DB: Database, Records: Copy> Clone for Outbox<DB, Records>

fn clone(&self) -> Self

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<DB: Database, Records: Debug> Debug for Outbox<DB, Records>

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

Formats the value using the given formatter.