1use crate::llm::actions::{
2 protocol_trait::{ActionResult, Protocol, Server},
3 ActionDefinition, Parameter, ParameterDefinition,
4};
5use crate::protocol::log_template::LogTemplate;
6use crate::protocol::{EventType, SpawnContext};
7use crate::state::app_state::AppState;
8use anyhow::Result;
9use serde_json::{json, Value};
10use std::sync::LazyLock;
11
12#[derive(Default)]
13pub struct ZenohProtocol;
14impl ZenohProtocol {
15 pub fn new() -> Self {
16 Self
17 }
18}
19
20pub fn parameter(name: &str, kind: &str, description: &str, required: bool) -> Parameter {
21 Parameter {
22 name: name.into(),
23 type_hint: kind.into(),
24 description: description.into(),
25 required,
26 }
27}
28
29pub fn action(
30 name: &str,
31 description: &str,
32 parameters: Vec<Parameter>,
33 example: Value,
34) -> ActionDefinition {
35 ActionDefinition {
36 name: name.into(),
37 description: description.into(),
38 parameters,
39 example,
40 log_template: Some(LogTemplate::new().with_info(format!("Zenoh {name}"))),
41 }
42}
43
44fn payload_params() -> Vec<Parameter> {
45 vec![
46 parameter(
47 "payload",
48 "string",
49 "The value as text, or hex when payload_encoding is hex",
50 true,
51 ),
52 parameter("payload_encoding", "string", "utf8 (default) or hex", false),
53 parameter(
54 "encoding",
55 "string",
56 "Zenoh encoding (media type), e.g. text/plain or application/json",
57 false,
58 ),
59 ]
60}
61
62pub fn put() -> ActionDefinition {
63 let mut p = vec![parameter(
64 "key",
65 "string",
66 "Key expression to publish on, e.g. demo/temp",
67 true,
68 )];
69 p.extend(payload_params());
70 action(
71 "zenoh_put",
72 "Publish a value on a key: every matching subscriber receives it",
73 p,
74 json!({"type": "zenoh_put", "key": "demo/temp", "payload": "21.5"}),
75 )
76}
77pub fn delete() -> ActionDefinition {
78 action(
79 "zenoh_delete",
80 "Publish a deletion on a key",
81 vec![parameter(
82 "key",
83 "string",
84 "Key expression to delete, e.g. demo/temp",
85 true,
86 )],
87 json!({"type": "zenoh_delete", "key": "demo/temp"}),
88 )
89}
90pub fn get() -> ActionDefinition {
91 action(
92 "zenoh_get",
93 "Query every matching queryable; the replies arrive as zenoh_get_result (10 s timeout, 256 replies)",
94 vec![parameter("selector", "string", "Key expression with optional ?parameters, e.g. demo/**?unit=c", true)],
95 json!({"type": "zenoh_get", "selector": "demo/**"}),
96 )
97}
98pub fn reply() -> ActionDefinition {
99 let mut p = vec![parameter(
100 "key",
101 "string",
102 "The reply's key (default: the query's key expression); must match the query",
103 false,
104 )];
105 p.extend(payload_params());
106 action(
107 "zenoh_reply",
108 "Answer the query with a value (several replies are allowed; none is an empty answer)",
109 p,
110 json!({"type": "zenoh_reply", "payload": "21.5"}),
111 )
112}
113pub fn reply_error() -> ActionDefinition {
114 action(
115 "zenoh_reply_error",
116 "Answer the query with an error the querier receives as such",
117 vec![
118 parameter(
119 "payload",
120 "string",
121 "The error text the querier receives, e.g. sensor offline",
122 true,
123 ),
124 parameter("payload_encoding", "string", "utf8 (default) or hex", false),
125 ],
126 json!({"type": "zenoh_reply_error", "payload": "sensor offline"}),
127 )
128}
129
130fn sample_params() -> Vec<Parameter> {
131 vec![
132 parameter("key", "string", "The key the value was published on", true),
133 parameter(
134 "kind",
135 "string",
136 "put when a value was published, delete when the key was deleted",
137 true,
138 ),
139 parameter(
140 "payload",
141 "string",
142 "The value as text, or hex per payload_encoding",
143 true,
144 ),
145 parameter(
146 "payload_encoding",
147 "string",
148 "utf8 when the payload is text, hex when it is not UTF-8",
149 true,
150 ),
151 parameter(
152 "encoding",
153 "string",
154 "The Zenoh encoding the publisher declared",
155 true,
156 ),
157 ]
158}
159
160pub static SAMPLE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
161 EventType::new("zenoh_sample", "A value (or deletion) arrived on a subscribed key expression; react with puts, deletes or gets, or nothing", json!({"type": "zenoh_put", "key": "demo/ack", "payload": "seen"}))
162 .with_parameters(sample_params())
163 .with_actions(vec![put(), delete(), get()])
164});
165pub static QUERY_EVENT: LazyLock<EventType> = LazyLock::new(|| {
166 EventType::new(
167 "zenoh_query",
168 "A query reached a declared queryable; answer with replies or an error",
169 reply().example.clone(),
170 )
171 .with_parameters(vec![
172 parameter("selector", "string", "The whole selector", true),
173 parameter("key", "string", "The queried key expression", true),
174 parameter(
175 "parameters",
176 "string",
177 "The selector's parameters after ?",
178 true,
179 ),
180 parameter("payload", "string", "A value sent with the query", false),
181 parameter(
182 "payload_encoding",
183 "string",
184 "utf8 when the payload is text, hex when it is not UTF-8",
185 false,
186 ),
187 ])
188 .with_actions(vec![reply(), reply_error(), put(), delete()])
189});
190pub static GET_RESULT_EVENT: LazyLock<EventType> = LazyLock::new(|| {
191 EventType::new(
192 "zenoh_get_result",
193 "The replies to a zenoh_get",
194 json!({"type": "zenoh_put", "key": "demo/summary", "payload": "done"}),
195 )
196 .with_parameters(vec![
197 parameter("selector", "string", "The selector queried", true),
198 parameter(
199 "replies",
200 "array",
201 "[{key, kind, payload, payload_encoding, encoding} | {error}]",
202 true,
203 ),
204 ])
205 .with_actions(vec![put(), delete(), get()])
206});
207
208pub fn key_list(name: &str, description: &str, example: Value) -> ParameterDefinition {
209 ParameterDefinition {
210 name: name.into(),
211 type_hint: "array".into(),
212 description: description.into(),
213 required: false,
214 example,
215 default: None,
216 }
217}
218
219impl Protocol for ZenohProtocol {
220 fn protocol_name(&self) -> &'static str {
221 "Zenoh"
222 }
223 fn stack_name(&self) -> &'static str {
224 "ETH>IP>TCP>Zenoh"
225 }
226 fn description(&self) -> &'static str {
227 "Zenoh router or peer listening on TCP: receives publications on subscribed keys and answers queries; the handler publishes, deletes and queries"
228 }
229 fn keywords(&self) -> Vec<&'static str> {
230 vec![
231 "zenoh",
232 "zenoh router",
233 "pub/sub",
234 "queryable",
235 "eclipse zenoh",
236 ]
237 }
238 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
239 vec![]
240 }
241 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
242 vec![put(), delete(), get(), reply(), reply_error()]
243 }
244 fn get_event_types(&self) -> Vec<EventType> {
245 vec![
246 SAMPLE_EVENT.clone(),
247 QUERY_EVENT.clone(),
248 GET_RESULT_EVENT.clone(),
249 ]
250 }
251 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
252 vec![
253 key_list(
254 "subscribe",
255 "Key expressions to subscribe to; each value raises zenoh_sample",
256 json!(["demo/**"]),
257 ),
258 key_list(
259 "queryable",
260 "Key expressions to answer queries on; each query raises zenoh_query",
261 json!(["demo/q/**"]),
262 ),
263 ParameterDefinition {
264 name: "mode".into(),
265 type_hint: "string".into(),
266 description: "router (routes between the clients that connect) or peer".into(),
267 required: false,
268 example: json!("peer"),
269 default: Some(json!(super::DEFAULT_MODE)),
270 },
271 ]
272 }
273 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
274 use crate::protocol::metadata::*;
275 ProtocolMetadataV2::builder()
276 .state(DevelopmentState::Experimental)
277 .privilege_requirement(PrivilegeRequirement::None)
278 .well_known_port(7447)
279 .implementation("The zenoh 1.10.1 runtime (TCP transport only, multicast scouting off) in router or peer mode; Rust declares the subscribers and queryables and runs the handler's actions; links are polled for the connection list")
280 .llm_control("Reactions to every publication and the answer to every query: puts, deletes, gets, replies and reply errors")
281 .e2e_testing("tests/server/zenoh: zenoh-pico 1.10.1 (independent C implementation) clients publish, subscribe, query and serve a queryable through NetGet")
282 .notes("No TLS, QUIC, UDP or shared memory transports; no storage (persist with memory or SQLite). Gets time out after 10 s with at most 256 replies; chains of handler actions stop at depth 4; 64 queries in flight.")
283 .request_only("Zenoh traffic is not per connection: the handler publishes, deletes and queries in answer to samples and queries")
284 .answers_on_failure()
285 .max_inbound_bytes(super::node::MAX_PAYLOAD)
286 .build()
287 }
288 fn example_prompt(&self) -> &'static str {
289 "Zenoh router on port 7447 that answers queries on demo/** with the current time"
290 }
291 fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
292 let llm = json!({"type":"open_server","base_stack":"zenoh","port":7447,"instruction":"Answer queries on demo/q/** with 42","startup_params":{"queryable":["demo/q/**"]}});
293 let mut static_example = llm.clone();
294 static_example["event_handlers"] = json!([{"event_pattern":"zenoh_query","handler":{"type":"static","actions":[{"type":"zenoh_reply","payload":"42"}]}}]);
295 let mut scripted = llm.clone();
296 scripted["event_handlers"] = json!([{"event_pattern":"zenoh_query","handler":{"type":"script","language":"python","code":"import json,sys\ne=json.load(sys.stdin)['event']\nprint(json.dumps({'actions':[{'type':'zenoh_reply','payload':'answer for '+e['key']}]}))"}}]);
297 crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
298 }
299 fn group_name(&self) -> &'static str {
300 "IoT"
301 }
302}
303
304impl Server for ZenohProtocol {
305 fn spawn(
306 &self,
307 ctx: SpawnContext,
308 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
309 {
310 Box::pin(super::spawn(ctx))
311 }
312 fn execute_action(&self, v: Value) -> Result<ActionResult> {
313 super::node::validate(&v)?;
314 Ok(ActionResult::Custom {
315 name: v["type"].as_str().unwrap_or_default().into(),
316 data: v,
317 })
318 }
319}