Shutdown

Shutdown unblocks every read and closes every connection; a publisher used afterwards gets RedisError::ShutDown at once.

When the service stops, the runtime runs the on_shutdown hooks while the broker is still connected and then shuts the broker down. A stream or list subscription may be waiting in a blocking read on its own connection. The broker sends CLIENT UNBLOCK for each such read, so the read ends at once instead of at its timeout, and closes the connection. Then the pool sends QUIT on every connection, and the after_shutdown hooks run.

A publisher taken before the shutdown keeps pointing at the closed connection. A publish through it, an acknowledgement of a delivery or a transaction returns RedisError::ShutDown, and never runs against the dead pool. A tool that drives the broker by hand receives a ClosedRedisBroker from the shutdown; its connections_closed is the number of pooled connections it closed.

Example

The service answers each order with a shipment, and logs once its connections are closed.

Macros

examples/redis_broker_shutdown.rs

#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            b.include(ship);
        })
        // Runs once every connection of the broker is closed.
        .after_shutdown(async move |_state| {
            tracing::info!("redis connections closed");
            Ok::<_, Infallible>(())
        })
}

Manual

examples/manual/redis_broker_shutdown.rs

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            b.include(
                subscriber(RedisStream::new("orders").group("workers"), Ship)
                    .reply()
                    .build(),
            );
        })
        // Runs once every connection of the broker is closed.
        .after_shutdown(async move |_state| {
            tracing::info!("redis connections closed");
            Ok::<_, Infallible>(())
        })
}