跳到主要内容
本页尚未翻译

下面显示的是英文原文。

Lifecycle

Four hooks run around the brokers' connection: on_startup, after_startup, on_shutdown and after_shutdown.

on_startup runs before any broker connects. The value it returns becomes the application state, and its error aborts startup with no broker dialled. on_startup comes first in the builder: a hook registered before it makes the builder panic. after_startup runs once every subscription is open; its error aborts startup and shuts down the brokers that connected. on_shutdown runs when shutdown begins, while the brokers are still connected. after_shutdown runs after the brokers have shut down. The shutdown hooks' errors are logged and do not stop the shutdown. Every hook after on_startup receives the state as Arc<State>. shutdown_timeout bounds how long shutdown waits for handlers still running; past it, they are aborted. Without it, shutdown waits for them indefinitely. A broker scope's after_startup receives a live publisher of that broker, paired from the policy it names. TestApp runs every hook in the service's order: the startup hooks on start, the shutdown hooks on shutdown.

Example​

examples/app_lifecycle.rs
/// A stand-in for a connection pool: it records what the service did with it.
struct Pool {
    log: Arc<Mutex<Vec<&'static str>>>,
}

impl Pool {
    fn open(log: Arc<Mutex<Vec<&'static str>>>) -> Self {
        let pool = Self { log };
        pool.record("open");
        pool
    }

    fn record(&self, event: &'static str) {
        if let Ok(mut log) = self.log.lock() {
            log.push(event);
        }
    }
}

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

fn app(log: Arc<Mutex<Vec<&'static str>>>) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Before the brokers connect: the value it produces is the application state.
        .on_startup(async move |()| Ok::<_, Infallible>(Pool::open(log)))
        // Once every subscription is open.
        .after_startup(async move |pool: Arc<Pool>| {
            pool.record("ready");
            Ok::<_, Infallible>(())
        })
        // When shutdown begins, while the brokers are still connected.
        .on_shutdown(async move |pool: Arc<Pool>| {
            pool.record("stopping");
            Ok::<_, Infallible>(())
        })
        // After the brokers have shut down.
        .after_shutdown(async move |pool: Arc<Pool>| {
            pool.record("closed");
            Ok::<_, Infallible>(())
        })
        // How long shutdown waits for handlers still running.
        .shutdown_timeout(Duration::from_secs(10))
        .with_broker(MemoryBroker::new(), |b| {
            b.include(accept);
        })
}

A hook of one broker​

examples/app_lifecycle.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "services.started")]
struct Started {
    service: String,
}

#[subscriber("orders")]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(accept);
        // Runs once the subscriptions are open, with a publisher of this broker.
        b.after_startup(Publish, async move |publisher| {
            let started = Started {
                service: "orders".to_owned(),
            };
            publisher.message(&started).publish().await
        });
    })
}