wrap

Outbox::wrap gives a publisher outside the publish pipeline the same tracking, for an HTTP endpoint or an after_startup hook.

The publish pipeline reaches what handlers publish. A publisher from RunningApp::publisher, or the one an after_startup hook receives, runs outside it, so the publish middleware never sees its messages. wrap takes such a publisher and returns a TrackedPublisher, a publisher like any other: a message under a registered name is recorded first, then sent with the record's id. A message under any other name goes straight to the wrapped publisher.

A publish through it fails with TrackedPublishError: Outbox when the record was not written and the message was not sent, Publish when the wrapped publisher failed after the record was written. In the second case the record stays unprocessed, and the next startup sends it. The wrapper borrows the payload for the record in whatever form the publisher takes it, so the message is not copied.

Example

The service starts its app by hand, takes the live publisher of a bound Publish policy, as an HTTP server would, and publishes each order through the wrapper.

Macros
Manual

examples/sqlx_outbox_wrap.rs
use ruststream::memory::prelude::*;
use ruststream_sqlx::outbox;
use sqlx::SqlitePool;

async fn serve(pool: SqlitePool, orders: &[u64]) -> Result<(), Box<dyn Error + Send + Sync>> {
    let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
    let broker = MemoryBroker::new().bindable();
    let egress = broker.bind(Publish);
    let app = RustStream::new(AppInfo::new("orders", "0.1.0"))
        .publish_layer(tracking.publish_layer())
        .with_broker(broker, |_b| {});
    let running = app.start().await?;

    // The running app's publisher bypasses the publish pipeline; `wrap` gives it the tracking.
    let publisher = tracking.wrap(running.publisher(egress).await?);
    for &id in orders {
        publisher.message(&OrderPlaced { id }).publish().await?;
        tracing::info!(order = id, "order recorded and sent");
    }

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

examples/manual/sqlx_outbox_wrap.rs
use ruststream::memory::prelude::*;
use ruststream_sqlx::outbox::Outbox;
use sqlx::SqlitePool;

async fn serve(pool: SqlitePool, orders: &[u64]) -> Result<(), Box<dyn Error + Send + Sync>> {
    let tracking = Outbox::new(pool).track::<OrderOutbox, Orders>();
    let broker = MemoryBroker::new().bindable();
    let egress = broker.bind(Publish);
    let app = RustStream::new(AppInfo::new("orders", "0.1.0"))
        .publish_layer(tracking.publish_layer())
        .with_broker(broker, |_b| {});
    let running = app.start().await?;

    // The running app's publisher bypasses the publish pipeline; `wrap` gives it the tracking.
    let publisher = tracking.wrap(running.publisher(egress).await?);
    for &id in orders {
        publisher.message(&OrderPlaced { id }).publish().await?;
        tracing::info!(order = id, "order recorded and sent");
    }

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

Example​

Example