set_pool

Inside #[ruststream::app] the registry starts without a pool, and on_startup builds one and hands it over with set_pool; the service closes it.

#[ruststream::app] builds the app before the Tokio runtime exists, and sqlx builds a pool only inside one. So the registry starts without a pool: outbox! without pool:, or Outbox::deferred by hand. on_startup runs inside the runtime, builds the pool and hands it to a clone of the registry with set_pool. Every handle the registry gave out, the middlewares and the republish among them, reads that one pool. An app built inside a runtime passes the pool directly instead: outbox! { pool: pool, .. } or Outbox::new.

set_pool takes one pool, once: a second call, or a call on a registry built with a pool, returns PoolAlreadySet. Until the pool is set, a tracked publish fails with OutboxError::NoPool, which names the name, and a tracked delivery is retried without its handler. The value on_startup returns becomes the app's state, and after_shutdown receives it: the service closes the pool there, once the handlers have finished and the brokers have stopped.

Example

service builds the app from a database URL, and #[ruststream::app] runs it on the service's own file.

Macros
Manual

examples/sqlx_outbox_pool.rs
use std::convert::Infallible;
use std::io;

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

fn service(url: String) -> impl App {
    // No pool yet: the Tokio runtime starts after this builder, and sqlx builds a pool only there.
    let tracking = outbox! { "orders" => OrderOutbox };
    let registry = tracking.clone();

    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .on_startup(async move |()| {
            let pool = SqlitePool::connect(&url).await.map_err(io::Error::other)?;
            // Every handle the registry gave out reads this one pool.
            registry.set_pool(pool.clone()).map_err(io::Error::other)?;
            Ok::<_, io::Error>(pool)
        })
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        // Runs once the handlers have finished and the brokers have stopped.
        .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());
        })
}

#[ruststream::app]
fn app() -> impl App {
    service("sqlite:orders.db".to_owned())
}

examples/manual/sqlx_outbox_pool.rs
use std::convert::Infallible;
use std::io;

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

fn service(url: String) -> impl App {
    // No pool yet: the Tokio runtime starts after this builder, and sqlx builds a pool only there.
    let tracking = Outbox::deferred().track::<OrderOutbox, Orders>();
    let registry = tracking.clone();

    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .on_startup(async move |()| {
            let pool = SqlitePool::connect(&url).await.map_err(io::Error::other)?;
            // Every handle the registry gave out reads this one pool.
            registry.set_pool(pool.clone()).map_err(io::Error::other)?;
            Ok::<_, io::Error>(pool)
        })
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        // Runs once the handlers have finished and the brokers have stopped.
        .after_shutdown(async move |pool| {
            pool.close().await;
            Ok::<_, Infallible>(())
        })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(subscriber("checkout", Place).reply().build())
                .out_reply(Publish);
            b.include(subscriber("orders", Fulfil).build());
            b.after_startup(Publish, tracking.republish());
        })
}

#[ruststream::app]
fn app() -> impl App {
    service("sqlite:orders.db".to_owned())
}

Example​

Example