Shutdown

Shutdown waits for the handlers in flight and drains the connection; a publisher used afterwards reports NatsError::Closed.

The runtime stops reading first and waits for the handlers that are still running. The app's shutdown_timeout bounds that wait; with none set it waits for every handler. Then the broker drains the connection: the client ends its subscriptions, sends what it still buffers and closes. So a reply or a publish a handler made before the shutdown reaches the server before the service exits. A drain that fails ends the shutdown with NatsError::Shutdown.

The broker marks the connection closed before it drains. A publisher taken earlier keeps pointing at the connection, and a publish through it returns NatsError::Closed with the subject it was asked to reach. It never sends into a connection that is going away.

A tool that drives the broker by hand receives a ClosedNatsBroker from the shutdown. It carries what the connection did over its life: messages_sent, messages_received and connects, reconnects included.

Example

Shutdown gives the handlers ten seconds, and the shipments they replied with reach the server before the service exits.

Macros

examples/nats_broker_shutdown.rs

/// The service on the servers at `addrs`; shutdown waits up to ten seconds for the handlers in
/// flight before it drains the connection.
fn service(addrs: &str) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .shutdown_timeout(Duration::from_secs(10))
        .with_broker(NatsBroker::new(addrs), |b| {
            b.include(ship);
        })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service("nats://localhost:4222")
}

Manual

examples/manual/nats_broker_shutdown.rs

/// The service on the servers at `addrs`; shutdown waits up to ten seconds for the handlers in
/// flight before it drains the connection.
fn service(addrs: &str) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .shutdown_timeout(Duration::from_secs(10))
        .with_broker(NatsBroker::new(addrs), |b| {
            b.include(subscriber("broker-shutdown.orders", Ship).reply().build());
        })
}

fn app() -> impl App<State = ()> {
    service("nats://localhost:4222")
}