Struct HealthProbe 

pub struct HealthProbe { /* private fields */ }

A cheap, cloneable handle reporting the service’s HealthState and how many of its subscriptions are consuming.

Obtained from RunningApp::health; backed by a watch channel written only on lifecycle transitions and when a subscription’s loop starts or ends, so reading is a lock-free borrow and nothing runs on the per-delivery hot path. The probe outlives shutdown (which consumes the app handle), so a sibling HTTP task keeps answering with the terminal state.

Watch semantics: observers always see the latest state; rapid transitions (ShuttingDown immediately followed by Stopped) may coalesce, so poll state for snapshots and use changed only as a wake-up.

Dropping the RunningApp without calling shutdown detaches the service (per the crate rule that destructors never block); a probe then keeps reporting the last observed state, and changed resolves only if a fail-fast failure still flips the state or a subscription still ends.

Implementations

impl HealthProbe

pub fn state(&self) -> HealthState

A snapshot of the current state.

pub fn is_running(&self) -> bool

Whether the service is currently running (the healthy answer for a liveness endpoint).

pub fn subscriptions(&self) -> SubscriptionCount

How many subscriptions are consuming, and how many the service started.

A subscription whose stream ended while the service runs leaves the active count, under the default SubscriptionEnd::Log. The lifecycle state stays Running: the service decides what a partly working service reports outside.

Examples

use axum::Router;
use axum::extract::State;
use axum::http::StatusCode;
use axum::routing::get;
use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, HealthProbe, RustStream};
use tokio::net::TcpListener;

/// Ready while every subscription the service started still consumes.
async fn readyz(State(health): State<HealthProbe>) -> StatusCode {
    let count = health.subscriptions();
    if health.is_running() && count.active == count.started {
        StatusCode::OK
    } else {
        StatusCode::SERVICE_UNAVAILABLE
    }
}

let app =
    RustStream::new(AppInfo::new("svc", "0.1.0")).register_broker(MemoryBroker::new());
let running = app.start().await?;
let routes = Router::new()
    .route("/readyz", get(readyz))
    .with_state(running.health());
axum::serve(TcpListener::bind("0.0.0.0:8080").await?, routes)
    .with_graceful_shutdown(running.stopping())
    .await?;
running.shutdown().await?;

pub async fn changed(&mut self) -> HealthState

Waits for the next change, a lifecycle transition or a subscription starting or ending, and returns the state.

If the service was detached (the app handle dropped without shutdown) and no further change can arrive, the future stays pending forever rather than spinning the caller’s loop; pair it with state when a snapshot is enough.

Cancel safety

Cancel-safe: dropping the future loses nothing; a fresh call observes the same channel.

Trait Implementations

impl Clone for HealthProbe

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for HealthProbe

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.