Shutdown

shutdown closes the connected form at once; a publisher handed out earlier reports ZmqError::NotConnected and never sends.

The crate keeps no queue of its own, so the broker has nothing to flush and its shutdown returns at once. The runtime first stops the subscriptions and lets the handlers in flight finish. Then it shuts each broker down, and a subscription's socket closes as the runtime drops it.

A publisher can outlive the broker: one a startup hook kept, one a sibling task took from RunningApp::publisher, or a clone of either. After shutdown, a publish through it returns ZmqError::NotConnected and sends nothing. An answer and a request after shutdown fail the same way.

Running in production covers what a service does at shutdown.

Example

The service announces on a fan-out that it has started. The hook's publisher lives as long as the hook; a task that kept it past shutdown would get ZmqError::NotConnected from its next publish.

Macros
Manual

examples/zmq_broker_shutdown.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        ZmqFanout::new(ZmqEndpoint::bind("tcp://0.0.0.0:5557")),
        |b| {
            // The publisher lives as long as the hook; a task that kept it past shutdown would
            // get `ZmqError::NotConnected` from it.
            b.after_startup(Publish, async move |publisher| {
                let started = Started {
                    service: "worker".to_owned(),
                };
                publisher.message(&started).publish().await
            });
        },
    )
}

examples/manual/zmq_broker_shutdown.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        ZmqFanout::new(ZmqEndpoint::bind("tcp://0.0.0.0:5557")),
        |b| {
            // The publisher lives as long as the hook; a task that kept it past shutdown would
            // get `ZmqError::NotConnected` from it.
            b.after_startup(Publish, async move |publisher| {
                let started = Started {
                    service: "worker".to_owned(),
                };
                publisher.message(&started).publish().await
            });
        },
    )
}

Example​

Example