Shutdown

Shutdown returns ClosedSqlxBroker, releases every advisory lock in work, and a publisher kept past it reports SqlxBrokerError::Closed.

At shutdown the runtime first stops the subscriptions and lets the handlers in work finish, up to its shutdown timeout. Then it shuts the broker down: the claim loops stop, and so does the extension of the leases. A claim the runtime cut short as it stopped a subscription may still run on the server; shutdown waits until the server has ended it and rolled its transaction back, so the stopped service holds no lock on the table. A delivery in work in the lease form keeps its lease, which runs out unless the delivery settles first; its row then returns to the queue. In the advisory lock form the broker releases the lock of every delivery in work and waits for the settlements in flight, so no lock of the broker is left.

A pool the broker built (from_url) closes last: it lends no new connection, and a connection a delivery in work still holds closes when it comes back. A settlement after the shutdown that needs a new connection, as in the lease form, then fails, and its row returns when its lease runs out. The service's own pool (new) stays open, and the service closes it after run returns. A handle taken before shutdown, such as a repository publisher, refuses every publish afterwards with SqlxBrokerError::Closed, and never writes a row past the shutdown. A tool that drives the broker by hand receives the ClosedSqlxBroker, whose locks_released and connections_closed count what the shutdown did to the advisory locks.

Example

run returns once the handlers have drained and the broker has shut down and closed the pool it built.

Macros
Manual

examples/sqlx_broker_shutdown.rs
/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts, and closes it
    // when the service stops, after the handlers have drained.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("exports", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<Export>("exports"), |b| {
            b.include(export);
        })
}

#[ruststream::app]
fn app() -> impl App {
    service("sqlite:exports.db?mode=rwc")
}

examples/manual/sqlx_broker_shutdown.rs
/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts, and closes it
    // when the service stops, after the handlers have drained.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("exports", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<Export>("exports"), |b| {
            let export = subscriber("exports", Exporter)
                .map_source(|_| InboxQueue::<Export>::new("exports"));
            b.include(export.build());
        })
}

#[ruststream::app]
fn app() -> impl App {
    service("sqlite:exports.db?mode=rwc")
}

Example​

Example