82 lines
2.7 KiB
Rust
82 lines
2.7 KiB
Rust
//! Feeds whether the swarm-queue client is connected into an OTEL gauge on a
|
|
//! timer. The client reconnects in the background and keeps running while
|
|
//! detached, so the gauge is the only place a detachment shows.
|
|
|
|
use std::time::Duration;
|
|
|
|
use async_nats::connection::State;
|
|
|
|
/// Half the exporter's one-minute interval, so every export carries a
|
|
/// value at most one tick old.
|
|
const TICK: Duration = Duration::from_secs(30);
|
|
|
|
/// Background loop: record the client's connection state every [`TICK`].
|
|
/// Detached task — runs for the harness's lifetime.
|
|
///
|
|
/// Records `0` until the first connect has produced a client, so an agent that
|
|
/// never reaches the queue reports disconnected instead of nothing.
|
|
pub async fn run() {
|
|
crate::otel_turn_metrics::record_swarm_queue_connected(false);
|
|
let client = crate::swarm_queue::client().await;
|
|
watch(
|
|
|| client.connection_state(),
|
|
crate::otel_turn_metrics::record_swarm_queue_connected,
|
|
TICK,
|
|
)
|
|
.await;
|
|
}
|
|
|
|
/// Whether `state` counts as connected. `Pending` (never connected) and
|
|
/// `Disconnected` both do not.
|
|
fn is_connected(state: &State) -> bool {
|
|
*state == State::Connected
|
|
}
|
|
|
|
/// Sample `state` every `tick` and hand `record` whether it is connected.
|
|
async fn watch(state: impl Fn() -> State, mut record: impl FnMut(bool), tick: Duration) {
|
|
loop {
|
|
record(is_connected(&state()));
|
|
tokio::time::sleep(tick).await;
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use std::sync::{Arc, Mutex};
|
|
use std::time::Duration;
|
|
|
|
use async_nats::connection::State;
|
|
|
|
use super::{is_connected, watch};
|
|
|
|
#[test]
|
|
fn only_a_connected_client_counts_as_connected() {
|
|
assert!(is_connected(&State::Connected));
|
|
assert!(!is_connected(&State::Disconnected));
|
|
assert!(!is_connected(&State::Pending));
|
|
}
|
|
|
|
#[tokio::test(start_paused = true)]
|
|
async fn the_gauge_follows_the_connection_state_as_it_changes() {
|
|
let state = Arc::new(Mutex::new(State::Connected));
|
|
let seen = Arc::new(Mutex::new(Vec::new()));
|
|
let task = tokio::spawn(watch(
|
|
{
|
|
let state = Arc::clone(&state);
|
|
move || state.lock().unwrap().clone()
|
|
},
|
|
{
|
|
let seen = Arc::clone(&seen);
|
|
move |connected| seen.lock().unwrap().push(connected)
|
|
},
|
|
Duration::from_secs(10),
|
|
));
|
|
tokio::time::sleep(Duration::from_secs(15)).await;
|
|
*state.lock().unwrap() = State::Disconnected;
|
|
tokio::time::sleep(Duration::from_secs(10)).await;
|
|
*state.lock().unwrap() = State::Connected;
|
|
tokio::time::sleep(Duration::from_secs(10)).await;
|
|
task.abort();
|
|
assert_eq!(*seen.lock().unwrap(), [true, true, false, true]);
|
|
}
|
|
}
|