zenkey_fleet/
serve.rs

1//! A mock responder for the dev loop (#121): stand up a queryable so a
2//! consumer can be tested — the explorers could ask and observe but never
3//! answer.
4//!
5//! The declaration discipline lives here, not in a frontend: one declared
6//! queryable, an acknowledged undeclare, and every incoming query surfaced
7//! as an owned view — the log of who asked *is* half the feature, doubling
8//! as a "who queries this key" probe.
9//!
10//! Deliberately no reply scripting: static bytes only. nuze and zsak own
11//! the embedded-language lane, at the cost of a Nushell dependency and a
12//! linked libpython; the shell covers dynamic cases by restarting the
13//! responder.
14
15use anyhow::{Result, anyhow};
16use zenoh::Session;
17use zenoh::handlers::FifoChannelHandler;
18
19/// One incoming query, owned — what the responder answered, and what a
20/// frontend logs.
21#[derive(Debug, Clone)]
22pub struct ServedQuery {
23    /// The query's full selector, verbatim (key expression + parameters).
24    pub selector: String,
25    /// The parameters half alone, verbatim ("" when none).
26    pub parameters: String,
27    /// The query body, when it carried one (refcounted, like a payload).
28    pub payload: Option<zenoh::bytes::ZBytes>,
29    /// The body's declared encoding, when said.
30    pub encoding: Option<String>,
31    /// The query's attachment, when it carried one (#117).
32    pub attachment: Option<zenoh::bytes::ZBytes>,
33}
34
35/// A declared queryable answering every query with one static body.
36pub struct MockResponder {
37    queryable: zenoh::query::Queryable<FifoChannelHandler<zenoh::query::Query>>,
38    reply: Vec<u8>,
39    encoding: Option<String>,
40}
41
42impl std::fmt::Debug for MockResponder {
43    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
44        f.debug_struct("MockResponder").finish_non_exhaustive()
45    }
46}
47
48/// Declare the responder on `keyexpr` (a full wire expression — explorers
49/// are un-namespaced). `complete` sets zenoh's completeness flag: a claim
50/// that this responder holds *all* the data the expression names — say it
51/// only when you mean it.
52pub async fn declare_responder(
53    session: &Session,
54    keyexpr: &str,
55    reply: Vec<u8>,
56    encoding: Option<&str>,
57    complete: bool,
58) -> Result<MockResponder> {
59    let queryable = session
60        .declare_queryable(keyexpr.to_string())
61        .complete(complete)
62        .await
63        .map_err(|e| anyhow!("declare queryable {keyexpr}: {e}"))?;
64    Ok(MockResponder {
65        queryable,
66        reply,
67        encoding: encoding.map(str::to_string),
68    })
69}
70
71impl MockResponder {
72    /// Answer the next query and return its view, or `None` once the
73    /// queryable is gone. The reply is addressed to the query's own key —
74    /// concrete where the query was concrete — and errors on the reply
75    /// path surface in the view's place at the caller's log, not silently.
76    pub async fn next(&self) -> Option<ServedQuery> {
77        let query = self.queryable.recv_async().await.ok()?;
78        let view = ServedQuery {
79            selector: query.selector().to_string(),
80            parameters: query.parameters().to_string(),
81            payload: query.payload().cloned(),
82            encoding: query.encoding().map(|e| e.to_string()),
83            attachment: query.attachment().cloned(),
84        };
85        let key = query.key_expr().clone();
86        let reply = query.reply(key, self.reply.clone());
87        let reply = match &self.encoding {
88            Some(e) => reply.encoding(e.as_str()),
89            None => reply,
90        };
91        // A failed reply is the *asker's* silence, not ours to hide — but
92        // the view still surfaces so the log records the ask.
93        let _ = reply.await;
94        Some(view)
95    }
96
97    /// Undeclare, acknowledged — like the write facade's publications.
98    pub async fn undeclare(self) -> Result<()> {
99        self.queryable
100            .undeclare()
101            .await
102            .map_err(|e| anyhow!("undeclare queryable: {e}"))
103    }
104}
zenkey_fleet/serve.rs

zenkey_fleet/
serve.rs
