republish

b.after_startup(policy, tracking.republish()) sends every unprocessed record again once the subscriptions are open; republish_names picks names.

Outbox::republish returns a Republishing, the hook after_startup runs with the broker's live publisher of the policy it names. For each registered name it reads the unprocessed records through the record's Recover event and publishes each one with its id, its payload and the headers its record kept. A consumer then takes each copy as any tracked delivery, so a copy whose record a consumer processed in the meantime is acknowledged without its handler. The hook runs once the subscriptions are open, so a consumer in the same service receives what its own service sent before a restart.

republish_names republishes only the names it lists, for a service whose names live on several brokers: each broker's after_startup republishes the names that broker carries. A name it lists that the registry does not hold panics when the app is built.

A record the republish cannot read fails startup with OutboxError::Recover, and one it cannot send with OutboxError::Republish. A service whose records stay with a selection of its own writes Recover itself. The republish allocates its body once per startup.

Example

The broker's after_startup republishes the unprocessed records of orders through its Publish policy; fulfil consumes them and marks each record.

Macros

examples/sqlx_outbox_republish.rs

use ruststream::memory::prelude::*;
use ruststream_sqlx::outbox;
use sqlx::SqlitePool;

fn service(pool: SqlitePool) -> impl App {
    let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
    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);
            // Runs once the subscriptions are open, through this broker's publisher.
            b.after_startup(Publish, tracking.republish());
        })
}

Manual

examples/manual/sqlx_outbox_republish.rs

use ruststream::memory::prelude::*;
use ruststream_sqlx::outbox::Outbox;
use sqlx::SqlitePool;

fn service(pool: SqlitePool) -> impl App {
    let tracking = Outbox::new(pool).track::<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(subscriber("checkout", Place).reply().build())
                .out_reply(Publish);
            b.include(subscriber("orders", Fulfil).build());
            // Runs once the subscriptions are open, through this broker's publisher.
            b.after_startup(Publish, tracking.republish());
        })
}