zenkey_fleet/bus/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 crate::{Error, Result};
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 /// Why the reply failed to send, when it did — `None` is a sent reply.
34 ///
35 /// Surfaced, not swallowed (deep-review D7): the ask itself is still a
36 /// fact worth logging either way, but a responder whose answers never
37 /// leave the process must say so, or its log reads as service
38 /// (RFC 05 §3.1 — silence needs attribution, on the answering side
39 /// too).
40 pub reply_error: Option<String>,
41}
42
43/// A declared queryable answering every query with one static body.
44pub struct MockResponder {
45 queryable: zenoh::query::Queryable<FifoChannelHandler<zenoh::query::Query>>,
46 /// The declared expression — the responder's own key when concrete
47 /// (RFC 05 §2.1: replies ride the responder's key, G-05b).
48 keyexpr: String,
49 /// Whether `keyexpr` is concrete (no wildcards) — decided once at
50 /// declaration.
51 concrete: bool,
52 reply: Vec<u8>,
53 encoding: Option<String>,
54}
55
56impl std::fmt::Debug for MockResponder {
57 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
58 f.debug_struct("MockResponder").finish_non_exhaustive()
59 }
60}
61
62/// Declare the responder on `keyexpr` (a full wire expression — explorers
63/// are un-namespaced). `complete` sets zenoh's completeness flag: a claim
64/// that this responder holds *all* the data the expression names — say it
65/// only when you mean it.
66pub async fn declare_responder(
67 session: &Session,
68 keyexpr: &str,
69 reply: Vec<u8>,
70 encoding: Option<&str>,
71 complete: bool,
72) -> Result<MockResponder> {
73 let parsed = zenoh::key_expr::KeyExpr::try_from(keyexpr.to_string())
74 .map_err(|e| Error::bus("declare queryable", keyexpr, e))?;
75 let concrete = !parsed.is_wild();
76 let queryable = crate::bus::teardown::declared(
77 "declare queryable",
78 keyexpr,
79 session.declare_queryable(parsed).complete(complete),
80 )
81 .await?;
82 Ok(MockResponder {
83 queryable,
84 keyexpr: keyexpr.to_string(),
85 concrete,
86 reply,
87 encoding: encoding.map(str::to_string),
88 })
89}
90
91impl MockResponder {
92 /// The next query, or `None` once the queryable is gone.
93 ///
94 /// **One await, and it consumes nothing it does not hand back** (#333).
95 /// Receiving and answering used to be a single `next()` with an await at
96 /// each end: a caller that raced it against anything — the moment a `--for`
97 /// deadline or a count timeout joins the `select!` that today holds only
98 /// `ctrl_c` — could be dropped between the two, and the query was then
99 /// taken off the channel, never answered, and never logged. Invisible on
100 /// both sides: the asker sees silence it cannot attribute (RFC 05 §3.1),
101 /// and the responder's log — half the point of this type — never records
102 /// the ask at all (RFC 13 §3 O6).
103 ///
104 /// The split is [`crate::bus::producer::Responder`]'s, right next door.
105 /// Hand what this returns to [`answer`](Self::answer): a query in the
106 /// caller's hand can still be answered after a cancelled poll, and one
107 /// never received was never taken.
108 pub async fn next(&self) -> Option<zenoh::query::Query> {
109 self.queryable.recv_async().await.ok()
110 }
111
112 /// The queries as a [`Stream`](futures_core::Stream) (#343).
113 ///
114 /// Borrows, which preserves #333's split exactly as [`next`](Self::next)
115 /// does: [`answer`](Self::answer) takes `&self`, so a query taken from
116 /// this stream is still answerable, and a stream dropped mid-poll has
117 /// consumed nothing it did not hand back.
118 pub fn stream(&self) -> impl futures_core::Stream<Item = zenoh::query::Query> + '_ {
119 self.queryable.stream()
120 }
121
122 /// Answer one query and return its view.
123 ///
124 /// The reply is addressed to the responder's **own declared key** when
125 /// that key is concrete (RFC 05 §2.1: attribution and consolidation both
126 /// read the reply key, so echoing a wildcard selector back — the G-05b
127 /// violation this used to commit — collapses a mocked fleet to one
128 /// surviving reply). A responder declared on a wildcard has no own
129 /// concrete key; it falls back to the query's key, which is concrete
130 /// exactly when the asker named a real key. An error on the reply path
131 /// rides the view ([`ServedQuery::reply_error`]) at the caller's log, not
132 /// silently.
133 pub async fn answer(&self, query: zenoh::query::Query) -> ServedQuery {
134 let mut view = ServedQuery {
135 selector: query.selector().to_string(),
136 parameters: query.parameters().to_string(),
137 payload: query.payload().cloned(),
138 encoding: query.encoding().map(|e| e.to_string()),
139 attachment: query.attachment().cloned(),
140 reply_error: None,
141 };
142 let key = if self.concrete {
143 self.keyexpr.clone()
144 } else {
145 query.key_expr().to_string()
146 };
147 let reply = query.reply(key, self.reply.clone());
148 let reply = match &self.encoding {
149 Some(e) => reply.encoding(e.as_str()),
150 None => reply,
151 };
152 // The view still surfaces on failure so the log records the ask —
153 // but a reply that never left carries its reason with it (D7).
154 view.reply_error = reply.await.err().map(|e| e.to_string());
155 view
156 }
157
158 /// Undeclare, acknowledged — like the write facade's publications.
159 pub async fn undeclare(self) -> Result<()> {
160 self.queryable
161 .undeclare()
162 .await
163 .map_err(|e| Error::bus("undeclare queryable", "", e))
164 }
165}