netget/client/zenoh/
actions.rs

1use crate::llm::actions::{
2    client_trait::{Client, ClientActionResult},
3    protocol_trait::Protocol,
4    ActionDefinition, ParameterDefinition,
5};
6use crate::protocol::{ConnectContext, EventType};
7use crate::server::zenoh::actions::{
8    action, delete, get, key_list, parameter, put, reply, reply_error,
9};
10use anyhow::{bail, Result};
11use serde_json::{json, Value};
12use std::sync::LazyLock;
13
14#[derive(Default)]
15pub struct ZenohClientProtocol;
16impl ZenohClientProtocol {
17    pub fn new() -> Self {
18        Self
19    }
20}
21
22fn disconnect() -> ActionDefinition {
23    action(
24        "disconnect",
25        "Close the Zenoh session",
26        vec![],
27        json!({"type": "disconnect"}),
28    )
29}
30
31pub static CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
32    EventType::new(
33        "zenoh_connected",
34        "The session is open; publish, query, or wait for samples and queries",
35        put().example.clone(),
36    )
37    .with_parameters(vec![
38        parameter("zid", "string", "This session's Zenoh id", true),
39        parameter(
40            "mode",
41            "string",
42            "The session mode this client opened: client (through a router) or peer",
43            true,
44        ),
45        parameter(
46            "links",
47            "array",
48            "The session's links: [{src, dst, zid}]",
49            true,
50        ),
51    ])
52    .with_actions(vec![put(), delete(), get(), disconnect()])
53});
54
55impl Protocol for ZenohClientProtocol {
56    fn protocol_name(&self) -> &'static str {
57        "Zenoh"
58    }
59    fn stack_name(&self) -> &'static str {
60        "ETH>IP>TCP>Zenoh"
61    }
62    fn keywords(&self) -> Vec<&'static str> {
63        vec!["zenoh", "zenoh client", "zenoh peer"]
64    }
65    fn description(&self) -> &'static str {
66        "Zenoh client or peer connecting over TCP: publishes, deletes and queries, and reacts to samples and queries on declared keys"
67    }
68    fn get_async_actions(&self, _: &crate::state::app_state::AppState) -> Vec<ActionDefinition> {
69        vec![put(), delete(), get(), disconnect()]
70    }
71    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
72        vec![reply(), reply_error()]
73    }
74    fn get_event_types(&self) -> Vec<EventType> {
75        let mut out = vec![CONNECTED_EVENT.clone()];
76        out.extend(crate::server::zenoh::events().into_iter().cloned());
77        out
78    }
79    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
80        vec![
81            key_list(
82                "subscribe",
83                "Key expressions to subscribe to; each value raises zenoh_sample",
84                json!(["demo/**"]),
85            ),
86            key_list(
87                "queryable",
88                "Key expressions to answer queries on; each query raises zenoh_query",
89                json!(["demo/q/**"]),
90            ),
91            ParameterDefinition {
92                name: "mode".into(),
93                type_hint: "string".into(),
94                description: "client (to a router) or peer".into(),
95                required: false,
96                example: json!("peer"),
97                default: Some(json!(super::DEFAULT_MODE)),
98            },
99        ]
100    }
101    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
102        use crate::protocol::metadata::*;
103        ProtocolMetadataV2::builder()
104            .state(DevelopmentState::Experimental)
105            .privilege_requirement(PrivilegeRequirement::None)
106            .implementation("The zenoh 1.10.1 runtime (TCP only, multicast scouting off) in client or peer mode connecting to remote_addr")
107            .llm_control("What to publish, delete and query, and how to react to samples and answer queries")
108            .e2e_testing("tests/client/zenoh: zenoh-pico 1.10.1 (independent C implementation) peers listening: one subscribes to what NetGet publishes, one serves a queryable NetGet queries, one publishes to NetGet's subscriber")
109            .notes("No TLS, QUIC or UDP transports. Gets time out after 10 s with at most 256 replies; chains stop at depth 4.")
110            .build()
111    }
112    fn example_prompt(&self) -> &'static str {
113        "Connect to the Zenoh router at 127.0.0.1:7447 and publish 21.5 on demo/temp"
114    }
115    fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
116        let llm = json!({"type":"open_client","protocol":"zenoh","remote_addr":"127.0.0.1:7447","instruction":"Publish 21.5 on demo/temp"});
117        let mut static_example = llm.clone();
118        static_example["event_handlers"] = json!([{"event_pattern":"zenoh_connected","handler":{"type":"static","actions":[put().example]}}]);
119        let mut scripted = llm.clone();
120        scripted["event_handlers"] = json!([{"event_pattern":"zenoh_connected","handler":{"type":"script","language":"python","code":"import json\nprint(json.dumps({'actions':[{'type':'zenoh_get','selector':'demo/**'}]}))"}}]);
121        crate::llm::actions::StartupExamples::new(llm, scripted, static_example)
122    }
123    fn group_name(&self) -> &'static str {
124        "IoT"
125    }
126}
127
128impl Client for ZenohClientProtocol {
129    fn connect(
130        &self,
131        ctx: ConnectContext,
132    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
133    {
134        Box::pin(super::connect(ctx))
135    }
136    fn execute_action(&self, v: Value) -> Result<ClientActionResult> {
137        match v["type"].as_str() {
138            Some("disconnect") => return Ok(ClientActionResult::Disconnect),
139            Some(
140                "zenoh_put" | "zenoh_delete" | "zenoh_get" | "zenoh_reply" | "zenoh_reply_error",
141            ) => crate::server::zenoh::node::validate(&v)?,
142            _ => bail!("Unknown Zenoh client action"),
143        }
144        Ok(ClientActionResult::Custom {
145            name: v["type"].as_str().unwrap_or_default().into(),
146            data: v,
147        })
148    }
149}
netget/client/zenoh/actions.rs

netget/client/zenoh/
actions.rs
