Shutdown

Shutdown ends the subscriptions, the publisher links and the connection in turn; a publisher used afterwards reports AmqpError::NotConnected.

When shutdown begins, the runtime stops reading and waits for the handlers still running. The app's shutdown_timeout bounds that wait; without it the runtime waits until every handler has finished. A publish returns once the broker has settled the message, so what the handlers published is on the broker by then, and nothing is left to send.

The broker then closes from the inside out: each subscription's session ends, the publishers' sender links detach, their shared session ends, and the connection closes. Each layer still has to carry the broker's answer to the layer inside it, which fixes the order. The connection closes only after every subscription's session has ended, the sessions of subscriptions still open at shutdown included, so a clean stop reports no error. Every step runs even after an earlier one fails, so a stuck link cannot leave the connection open. The shutdown reports the first failure in that order: AmqpError::Detach for a link, AmqpError::Session for the session, AmqpError::Connect for the connection.

A publisher taken before the shutdown keeps pointing at the closed connection. A publish through it, or a transaction begun on a transactional publisher, returns AmqpError::NotConnected and never succeeds against the dead connection. A delivery settled after the shutdown reports an error, and the stream of a subscription that was still open ends.

Example

Handlers still running at shutdown get ten seconds to finish before the broker closes.

Macros
Manual

examples/amqp_broker_shutdown.rs
/// The service on `broker`. Once shutdown begins, the handlers still running get ten seconds to
/// finish before the broker closes.
fn service(broker: AmqpBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(broker, |b| {
            b.include(place);
        })
        .shutdown_timeout(Duration::from_secs(10))
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(AmqpBroker::new("amqp://artemis:artemis@localhost:5672"))
}

examples/manual/amqp_broker_shutdown.rs
/// The service on `broker`. Once shutdown begins, the handlers still running get ten seconds to
/// finish before the broker closes.
fn service(broker: AmqpBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(broker, |b| {
            let place = subscriber(AmqpAddress::queue(orders()), Place);
            b.include(place.build());
        })
        .shutdown_timeout(Duration::from_secs(10))
}

fn app() -> impl App<State = ()> {
    service(AmqpBroker::new("amqp://artemis:artemis@localhost:5672"))
}

Example​

Example