dora_message/
dynamic_node.rs

1//! The dynamic-node configuration reply uses JSON, unlike ordinary node IPC.
2//!
3//! Keep its optional bootstrap information outside `DaemonReply` and
4//! `NodeConfig`: their positional binary representations must not change.
5//! The `NodeConfig` variant and `result` retain the legacy JSON shape. Old
6//! clients ignore `zenoh`; new clients default it when an old daemon replies.
7
8use crate::daemon_to_node::NodeConfig;
9
10#[derive(Debug, serde::Serialize, serde::Deserialize)]
11pub enum DynamicNodeConfigReply {
12    NodeConfig {
13        result: Result<NodeConfig, String>,
14        #[serde(default, skip_serializing_if = "Option::is_none")]
15        zenoh: Option<DynamicNodePeering>,
16    },
17}
18
19/// A dynamic node's listener and the local graph neighbours it must dial.
20#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
21pub struct DynamicNodePeering {
22    pub listen: String,
23    pub connect: Vec<String>,
24}
25
26#[cfg(test)]
27mod tests {
28    use super::*;
29    use crate::daemon_to_node::DaemonReply;
30
31    fn config() -> NodeConfig {
32        NodeConfig {
33            dataflow_id: uuid::Uuid::nil(),
34            node_id: "dynamic".to_owned().into(),
35            run_config: Default::default(),
36            daemon_communication: None,
37            dataflow_descriptor: serde_yaml::Value::Null,
38            dynamic: true,
39            write_events_to: None,
40            restart_count: 0,
41            output_routing: None,
42        }
43    }
44
45    #[test]
46    fn old_client_accepts_reply_with_peering() {
47        let reply = DynamicNodeConfigReply::NodeConfig {
48            result: Ok(config()),
49            zenoh: Some(DynamicNodePeering {
50                listen: "tcp/127.0.0.1:1234".into(),
51                connect: vec!["tcp/127.0.0.1:1235".into()],
52            }),
53        };
54        let encoded = serde_json::to_vec(&reply).unwrap();
55        let old: DaemonReply = serde_json::from_slice(&encoded).unwrap();
56        assert!(matches!(old, DaemonReply::NodeConfig { result: Ok(c) } if c.dynamic));
57        let DynamicNodeConfigReply::NodeConfig { zenoh, .. } =
58            serde_json::from_slice(&encoded).unwrap();
59        assert_eq!(zenoh.unwrap().connect, ["tcp/127.0.0.1:1235"]);
60    }
61
62    #[test]
63    fn new_client_accepts_legacy_success_and_error() {
64        for result in [Ok(config()), Err("no node with ID `dynamic`".into())] {
65            let success = result.is_ok();
66            let encoded = serde_json::to_vec(&DaemonReply::NodeConfig { result }).unwrap();
67            let DynamicNodeConfigReply::NodeConfig { result, zenoh } =
68                serde_json::from_slice(&encoded).unwrap();
69            assert_eq!(result.is_ok(), success);
70            assert!(zenoh.is_none());
71        }
72    }
73}
dora_message/dynamic_node.rs

dora_message/
dynamic_node.rs
