zenkey_fleet/
admin.rs

1//! Zenoh admin-space access (issue #14): browse `@/**` — the middleware's
2//! own introspection — from the same un-namespaced session the convention
3//! tooling already holds (a namespaced session's admin selector would be
4//! rewritten and match nothing, RFC 09 §5).
5//!
6//! Admin key layouts vary between zenoh versions (report §3.1's caveat), so
7//! this module stays a thin, honest transport: keys + JSON values, no
8//! hardcoded schema. `routers` extracts the few fields every 1.x layout
9//! carries, and leaves the rest visible in `raw`.
10
11use std::time::Duration;
12
13use anyhow::{Result, anyhow};
14use zenoh::Session;
15use zenoh::query::{ConsolidationMode, QueryTarget};
16
17/// One admin-space entry.
18#[derive(Debug, Clone)]
19pub struct AdminEntry {
20    pub key: String,
21    pub value: serde_json::Value,
22}
23
24/// GET an admin selector (default `@/**`). Fans to every node (target All,
25/// consolidation None — several routers may answer).
26pub async fn admin_get(
27    session: &Session,
28    selector: &str,
29    timeout: Duration,
30) -> Result<Vec<AdminEntry>> {
31    let replies = session
32        .get(selector)
33        .target(QueryTarget::All)
34        .consolidation(ConsolidationMode::None)
35        .timeout(timeout)
36        .await
37        .map_err(|e| anyhow!("admin get {selector}: {e}"))?;
38    let mut out = Vec::new();
39    while let Ok(reply) = replies.recv_async().await {
40        let Ok(sample) = reply.result() else { continue };
41        let bytes = sample.payload().to_bytes();
42        let value = serde_json::from_slice(&bytes).unwrap_or_else(|_| {
43            serde_json::Value::String(String::from_utf8_lossy(&bytes).to_string())
44        });
45        out.push(AdminEntry {
46            key: sample.key_expr().as_str().to_string(),
47            value,
48        });
49    }
50    out.sort_by(|a, b| a.key.cmp(&b.key));
51    Ok(out)
52}
53
54/// A router (or peer) as the admin space reports it.
55#[derive(Debug, Clone, serde::Serialize)]
56pub struct RouterInfo {
57    pub zid: String,
58    #[serde(skip_serializing_if = "Option::is_none")]
59    pub version: Option<String>,
60    #[serde(skip_serializing_if = "Vec::is_empty")]
61    pub locators: Vec<String>,
62    /// The full admin document, untrimmed — layouts vary by version.
63    pub raw: serde_json::Value,
64}
65
66/// Enumerate routers/peers from `@/*/router` (and the fields every layout
67/// carries).
68pub async fn routers(session: &Session, timeout: Duration) -> Result<Vec<RouterInfo>> {
69    let entries = admin_get(session, "@/*/router", timeout).await?;
70    Ok(entries
71        .into_iter()
72        .map(|e| {
73            let zid = e
74                .value
75                .get("zid")
76                .and_then(|v| v.as_str())
77                .map(str::to_string)
78                .unwrap_or_else(|| {
79                    // Fall back to the key's zid chunk: @/<zid>/router.
80                    e.key.split('/').nth(1).unwrap_or("?").to_string()
81                });
82            let version = e
83                .value
84                .get("version")
85                .and_then(|v| v.as_str())
86                .map(str::to_string);
87            let locators = e
88                .value
89                .get("locators")
90                .and_then(|v| v.as_array())
91                .map(|a| {
92                    a.iter()
93                        .filter_map(|l| l.as_str().map(str::to_string))
94                        .collect()
95                })
96                .unwrap_or_default();
97            RouterInfo {
98                zid,
99                version,
100                locators,
101                raw: e.value,
102            }
103        })
104        .collect())
105}
zenkey_fleet/admin.rs

zenkey_fleet/
admin.rs
