zenoh_config/
lib.rs

1//
2// Copyright (c) 2023 ZettaScale Technology
3//
4// This program and the accompanying materials are made available under the
5// terms of the Eclipse Public License 2.0 which is available at
6// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
7// which is available at https://www.apache.org/licenses/LICENSE-2.0.
8//
9// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
10//
11// Contributors:
12//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
13//
14
15//! ⚠️ WARNING ⚠️
16//!
17//! This crate is intended for Zenoh's internal use.
18//!
19//! [Click here for Zenoh's documentation](https://docs.rs/zenoh/latest/zenoh)
20//!
21//! Configuration to pass to `zenoh::open()` and `zenoh::scout()` functions and associated constants.
22#![allow(deprecated)]
23
24pub mod defaults;
25mod include;
26pub mod qos;
27pub mod wrappers;
28
29#[allow(unused_imports)]
30use std::convert::TryFrom;
31// This is a false positive from the rust analyser
32use std::{
33    any::Any,
34    collections::HashSet,
35    fmt,
36    io::Read,
37    net::SocketAddr,
38    num::{NonZeroU16, NonZeroUsize},
39    ops::{self, Bound, Deref, RangeBounds},
40    path::Path,
41    sync::{Arc, Weak},
42};
43
44use include::recursive_include;
45use nonempty_collections::NEVec;
46use qos::{PublisherQoSConfList, QosFilter, QosOverwriteMessage, QosOverwrites};
47use secrecy::{CloneableSecret, DebugSecret, Secret, SerializableSecret, Zeroize};
48use serde::{Deserialize, Serialize};
49use serde_json::{Map, Value};
50use validated_struct::ValidatedMapAssociatedTypes;
51pub use validated_struct::{GetError, ValidatedMap};
52pub use wrappers::ZenohId;
53pub use zenoh_protocol::core::{
54    whatami, EndPoint, Locator, WhatAmI, WhatAmIMatcher, WhatAmIMatcherVisitor,
55};
56use zenoh_protocol::{
57    core::{
58        key_expr::{OwnedKeyExpr, OwnedNonWildKeyExpr},
59        Bits,
60    },
61    transport::{BatchSize, TransportSn},
62};
63use zenoh_result::{bail, zerror, ZResult};
64use zenoh_util::{LibLoader, LibSearchDirs};
65
66pub mod mode_dependent;
67pub use mode_dependent::*;
68
69pub mod connection_retry;
70pub use connection_retry::*;
71
72// Wrappers for secrecy of values
73#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
74pub struct SecretString(String);
75
76impl ops::Deref for SecretString {
77    type Target = String;
78
79    fn deref(&self) -> &Self::Target {
80        &self.0
81    }
82}
83
84impl SerializableSecret for SecretString {}
85impl DebugSecret for SecretString {}
86impl CloneableSecret for SecretString {}
87impl Zeroize for SecretString {
88    fn zeroize(&mut self) {
89        self.0 = "".to_string();
90    }
91}
92
93pub type SecretValue = Secret<SecretString>;
94
95#[derive(Debug, Deserialize, Serialize, Clone)]
96pub struct TransportWeight {
97    /// A zid of destination node.
98    pub dst_zid: ZenohId,
99    /// A weight of link from this node to the destination.
100    pub weight: NonZeroU16,
101}
102
103#[derive(Debug, Deserialize, Serialize, Clone, Copy, Eq, PartialEq)]
104#[serde(rename_all = "snake_case")]
105pub enum InterceptorFlow {
106    Egress,
107    Ingress,
108}
109
110#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, PartialEq)]
111#[serde(rename_all = "snake_case")]
112pub enum DownsamplingMessage {
113    Delete,
114    #[deprecated = "Use `Put` or `Delete` instead."]
115    Push,
116    Put,
117    Query,
118    Reply,
119}
120
121#[derive(Debug, Deserialize, Serialize, Clone)]
122#[serde(deny_unknown_fields)]
123pub struct DownsamplingRuleConf {
124    /// A list of key-expressions to which the downsampling will be applied.
125    /// Downsampling will be applied for all key extensions if the parameter is None
126    pub key_expr: OwnedKeyExpr,
127    /// The maximum frequency in Hertz;
128    pub freq: f64,
129}
130
131#[derive(Debug, Deserialize, Serialize, Clone)]
132#[serde(deny_unknown_fields)]
133pub struct DownsamplingItemConf {
134    /// Optional identifier for the downsampling configuration item
135    pub id: Option<String>,
136    /// A list of interfaces to which the downsampling will be applied
137    /// Downsampling will be applied for all interfaces if the parameter is None
138    pub interfaces: Option<NEVec<String>>,
139    /// A list of link types, transports having one of those link types will have the downsampling applied
140    /// Downsampling will be applied for all link types if the parameter is None
141    pub link_protocols: Option<NEVec<InterceptorLink>>,
142    // list of message types on which the downsampling will be applied
143    pub messages: NEVec<DownsamplingMessage>,
144    /// A list of downsampling rules: key_expression and the maximum frequency in Hertz
145    pub rules: NEVec<DownsamplingRuleConf>,
146    /// Downsampling flow directions: egress and/or ingress
147    pub flows: Option<NEVec<InterceptorFlow>>,
148}
149
150fn downsampling_validator(d: &Vec<DownsamplingItemConf>) -> bool {
151    for item in d {
152        if item
153            .messages
154            .iter()
155            .any(|m| *m == DownsamplingMessage::Push)
156        {
157            tracing::warn!("In 'downsampling/messages' configuration: 'push' is deprecated and may not be supported in future versions, use 'put' and/or 'delete' instead");
158        }
159    }
160    true
161}
162
163#[derive(Serialize, Debug, Deserialize, Clone)]
164#[serde(deny_unknown_fields)]
165pub struct LowPassFilterConf {
166    pub id: Option<String>,
167    pub interfaces: Option<NEVec<String>>,
168    pub link_protocols: Option<NEVec<InterceptorLink>>,
169    pub flows: Option<NEVec<InterceptorFlow>>,
170    pub messages: NEVec<LowPassFilterMessage>,
171    pub key_exprs: NEVec<OwnedKeyExpr>,
172    pub size_limit: usize,
173}
174
175#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, Hash, PartialEq)]
176#[serde(rename_all = "snake_case")]
177pub enum LowPassFilterMessage {
178    Put,
179    Delete,
180    Query,
181    Reply,
182}
183
184#[derive(Serialize, Debug, Deserialize, Clone)]
185#[serde(deny_unknown_fields)]
186pub struct AclConfigRule {
187    pub id: String,
188    pub key_exprs: NEVec<OwnedKeyExpr>,
189    pub messages: NEVec<AclMessage>,
190    pub flows: Option<NEVec<InterceptorFlow>>,
191    pub permission: Permission,
192}
193
194#[derive(Serialize, Debug, Deserialize, Clone)]
195#[serde(deny_unknown_fields)]
196pub struct AclConfigSubjects {
197    pub id: String,
198    pub interfaces: Option<NEVec<Interface>>,
199    pub cert_common_names: Option<NEVec<CertCommonName>>,
200    pub usernames: Option<NEVec<Username>>,
201    pub link_protocols: Option<NEVec<InterceptorLink>>,
202    pub zids: Option<NEVec<ZenohId>>,
203}
204
205#[derive(Debug, Clone, PartialEq, Eq)]
206pub struct ConfRange {
207    start: Option<u64>,
208    end: Option<u64>,
209}
210
211impl ConfRange {
212    pub fn new(start: Option<u64>, end: Option<u64>) -> Self {
213        Self { start, end }
214    }
215}
216
217impl RangeBounds<u64> for ConfRange {
218    fn start_bound(&self) -> Bound<&u64> {
219        match self.start {
220            Some(ref start) => Bound::Included(start),
221            None => Bound::Unbounded,
222        }
223    }
224    fn end_bound(&self) -> Bound<&u64> {
225        match self.end {
226            Some(ref end) => Bound::Included(end),
227            None => Bound::Unbounded,
228        }
229    }
230}
231
232impl serde::Serialize for ConfRange {
233    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
234    where
235        S: serde::Serializer,
236    {
237        serializer.serialize_str(&format!(
238            "{}..{}",
239            self.start.unwrap_or_default(),
240            self.end.unwrap_or_default()
241        ))
242    }
243}
244
245impl<'a> serde::Deserialize<'a> for ConfRange {
246    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
247    where
248        D: serde::Deserializer<'a>,
249    {
250        struct V;
251
252        impl<'de> serde::de::Visitor<'de> for V {
253            type Value = ConfRange;
254
255            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
256                formatter.write_str("range string")
257            }
258
259            fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
260            where
261                E: serde::de::Error,
262            {
263                let (start, end) = v
264                    .split_once("..")
265                    .ok_or_else(|| serde::de::Error::custom("invalid range"))?;
266                let parse_bound = |bound: &str| {
267                    (!bound.is_empty())
268                        .then(|| bound.parse::<u64>())
269                        .transpose()
270                        .map_err(|_| serde::de::Error::custom("invalid range bound"))
271                };
272                Ok(ConfRange::new(parse_bound(start)?, parse_bound(end)?))
273            }
274        }
275        deserializer.deserialize_str(V)
276    }
277}
278
279#[derive(Debug, Deserialize, Serialize, Clone)]
280#[serde(deny_unknown_fields)]
281pub struct QosOverwriteItemConf {
282    /// Optional identifier for the qos modification configuration item.
283    pub id: Option<String>,
284    /// A list of ZIDs on which qos will be overwritten when communicating with.
285    pub zids: Option<NEVec<ZenohId>>,
286    /// A list of interfaces to which the qos will be applied.
287    /// QosOverwrite will be applied for all interfaces if the parameter is None.
288    pub interfaces: Option<NEVec<String>>,
289    /// A list of link types, transports having one of those link types will have the qos overwrite applied
290    /// Qos overwrite will be applied for all link types if the parameter is None.
291    pub link_protocols: Option<NEVec<InterceptorLink>>,
292    /// List of message types on which the qos overwrite will be applied.
293    pub messages: NEVec<QosOverwriteMessage>,
294    /// List of key expressions to apply qos overwrite.
295    pub key_exprs: Option<NEVec<OwnedKeyExpr>>,
296    // The qos value to overwrite with.
297    pub overwrite: QosOverwrites,
298    /// QosOverwrite flow directions: egress and/or ingress.
299    pub flows: Option<NEVec<InterceptorFlow>>,
300    /// QoS filter to apply to the messages matching this item.
301    pub qos: Option<QosFilter>,
302    /// payload_size range for the messages matching this item.
303    pub payload_size: Option<ConfRange>,
304}
305
306#[derive(Serialize, Debug, Deserialize, Clone, PartialEq, Eq, Hash)]
307pub struct Interface(pub String);
308
309impl std::fmt::Display for Interface {
310    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
311        write!(f, "Interface({})", self.0)
312    }
313}
314
315#[derive(Serialize, Debug, Deserialize, Clone, PartialEq, Eq, Hash)]
316pub struct CertCommonName(pub String);
317
318impl std::fmt::Display for CertCommonName {
319    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
320        write!(f, "CertCommonName({})", self.0)
321    }
322}
323
324#[derive(Serialize, Debug, Deserialize, Clone, PartialEq, Eq, Hash)]
325pub struct Username(pub String);
326
327impl std::fmt::Display for Username {
328    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
329        write!(f, "Username({})", self.0)
330    }
331}
332
333#[derive(Serialize, Debug, Deserialize, Clone, PartialEq, Eq, Hash)]
334#[serde(rename_all = "kebab-case")]
335pub enum InterceptorLink {
336    Tcp,
337    Udp,
338    Tls,
339    Quic,
340    Serial,
341    Unixpipe,
342    UnixsockStream,
343    Vsock,
344    Ws,
345}
346
347impl std::fmt::Display for InterceptorLink {
348    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
349        write!(f, "Transport({self:?})")
350    }
351}
352
353#[derive(Serialize, Debug, Deserialize, Clone, PartialEq, Eq, Hash)]
354#[serde(deny_unknown_fields)]
355pub struct AclConfigPolicyEntry {
356    pub id: Option<String>,
357    pub rules: Vec<String>,
358    pub subjects: Vec<String>,
359}
360
361#[derive(Clone, Serialize, Debug, Deserialize)]
362#[serde(deny_unknown_fields)]
363pub struct PolicyRule {
364    pub subject_id: usize,
365    pub key_expr: OwnedKeyExpr,
366    pub message: AclMessage,
367    pub permission: Permission,
368    pub flow: InterceptorFlow,
369}
370
371#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, Hash, PartialEq)]
372#[serde(rename_all = "snake_case")]
373pub enum AclMessage {
374    Put,
375    Delete,
376    DeclareSubscriber,
377    Query,
378    DeclareQueryable,
379    Reply,
380    LivelinessToken,
381    DeclareLivelinessSubscriber,
382    LivelinessQuery,
383}
384
385#[derive(Clone, Copy, Debug, Serialize, Deserialize, Eq, Hash, PartialEq)]
386#[serde(rename_all = "snake_case")]
387pub enum Permission {
388    Allow,
389    Deny,
390}
391
392/// Strategy for autoconnection, mainly to avoid nodes connecting to each other redundantly.
393#[derive(Default, Clone, Copy, Debug, Serialize, Deserialize, Eq, Hash, PartialEq)]
394#[serde(rename_all = "kebab-case")]
395pub enum AutoConnectStrategy {
396    /// Always attempt to connect to another node, may result in redundant connection which
397    /// will be then be closed.
398    #[default]
399    Always,
400    /// A node will attempt to connect to another one only if its own zid is greater than the
401    /// other one. If both nodes use this strategy, only one will attempt the connection.
402    /// This strategy may not be suited if one of the node is not reachable by the other one,
403    /// for example because of a private IP.
404    GreaterZid,
405}
406
407pub trait ConfigValidator: Send + Sync {
408    fn check_config(
409        &self,
410        _plugin_name: &str,
411        _path: &str,
412        _current: &serde_json::Map<String, serde_json::Value>,
413        _new: &serde_json::Map<String, serde_json::Value>,
414    ) -> ZResult<Option<serde_json::Map<String, serde_json::Value>>> {
415        Ok(None)
416    }
417}
418
419// Necessary to allow to set default emplty weak reference value to plugin.validator field
420// because empty weak value is not allowed for Arc<dyn Trait>
421impl ConfigValidator for () {}
422
423/// Creates an empty zenoh net Session configuration.
424pub fn empty() -> Config {
425    Config::default()
426}
427
428/// Creates a default zenoh net Session configuration (equivalent to `peer`).
429pub fn default() -> Config {
430    peer()
431}
432
433/// Creates a default `'peer'` mode zenoh net Session configuration.
434pub fn peer() -> Config {
435    let mut config = Config::default();
436    config.set_mode(Some(WhatAmI::Peer)).unwrap();
437    config
438}
439
440/// Creates a default `'client'` mode zenoh net Session configuration.
441pub fn client<I: IntoIterator<Item = T>, T: Into<EndPoint>>(peers: I) -> Config {
442    let mut config = Config::default();
443    config.set_mode(Some(WhatAmI::Client)).unwrap();
444    config.connect.endpoints =
445        ModeDependentValue::Unique(peers.into_iter().map(|t| t.into()).collect());
446    config
447}
448
449#[test]
450fn config_keys() {
451    let c = Config::default();
452    dbg!(Vec::from_iter(c.keys()));
453}
454
455validated_struct::validator! {
456    #[derive(Default)]
457    #[recursive_attrs]
458    #[derive(serde::Deserialize, serde::Serialize, Clone, Debug)]
459    #[serde(default)]
460    #[serde(deny_unknown_fields)]
461    #[doc(hidden)]
462    Config {
463        /// The Zenoh ID of the instance. This ID MUST be unique throughout your Zenoh infrastructure and cannot exceed 16 bytes of length. If left unset, a random u128 will be generated.
464        /// If not specified a random Zenoh ID will be generated upon session creation.
465        id: Option<ZenohId>,
466        /// The metadata of the instance. Arbitrary json data available from the admin space
467        metadata: Value,
468        /// The node's mode ("router" (default value in `zenohd`), "peer" or "client").
469        mode: Option<whatami::WhatAmI>,
470        /// Which zenoh nodes to connect to.
471        pub connect:
472        ConnectConfig {
473            /// global timeout for full connect cycle
474            pub timeout_ms: Option<ModeDependentValue<i64>>,
475            /// The list of endpoints to connect to
476            pub endpoints: ModeDependentValue<Vec<EndPoint>>,
477            /// if connection timeout exceed, exit from application
478            pub exit_on_failure: Option<ModeDependentValue<bool>>,
479            pub retry: Option<connection_retry::ConnectionRetryModeDependentConf>,
480        },
481        /// Which endpoints to listen on.
482        pub listen:
483        ListenConfig {
484            /// global timeout for full listen cycle
485            pub timeout_ms: Option<ModeDependentValue<i64>>,
486            /// The list of endpoints to listen on
487            pub endpoints: ModeDependentValue<Vec<EndPoint>>,
488            /// if connection timeout exceed, exit from application
489            pub exit_on_failure: Option<ModeDependentValue<bool>>,
490            pub retry: Option<connection_retry::ConnectionRetryModeDependentConf>,
491        },
492        /// Configure the session open behavior.
493        pub open: #[derive(Default)]
494        OpenConf {
495            /// Configure the conditions to be met before session open returns.
496            pub return_conditions: #[derive(Default)]
497            ReturnConditionsConf {
498                /// Session open waits to connect to scouted peers and routers before returning.
499                /// When set to false, first publications and queries after session open from peers may be lost.
500                connect_scouted: Option<bool>,
501                /// Session open waits to receive initial declares from connected peers before returning.
502                /// Setting to false may cause extra traffic at startup from peers.
503                declares: Option<bool>,
504            },
505        },
506        pub scouting: #[derive(Default)]
507        ScoutingConf {
508            /// In client mode, the period dedicated to scouting for a router before failing. In milliseconds.
509            timeout: Option<u64>,
510            /// In peer mode, the period dedicated to scouting remote peers before attempting other operations. In milliseconds.
511            delay: Option<u64>,
512            /// The multicast scouting configuration.
513            pub multicast: #[derive(Default)]
514            ScoutingMulticastConf {
515                /// Whether multicast scouting is enabled or not. If left empty, `zenohd` will set it according to the presence of the `--no-multicast-scouting` argument.
516                enabled: Option<bool>,
517                /// The socket which should be used for multicast scouting. `zenohd` will use `224.0.0.224:7446` by default if none is provided.
518                address: Option<SocketAddr>,
519                /// The network interface which should be used for multicast scouting. `zenohd` will automatically select an interface if none is provided.
520                interface: Option<String>,
521                /// The time-to-live on multicast scouting packets. (default: 1)
522                pub ttl: Option<u32>,
523                /// Which type of Zenoh instances to automatically establish sessions with upon discovery through UDP multicast.
524                autoconnect: Option<ModeDependentValue<WhatAmIMatcher>>,
525                /// Strategy for autoconnection, mainly to avoid nodes connecting to each other redundantly.
526                autoconnect_strategy: Option<ModeDependentValue<TargetDependentValue<AutoConnectStrategy>>>,
527                /// Whether or not to listen for scout messages on UDP multicast and reply to them.
528                listen: Option<ModeDependentValue<bool>>,
529            },
530            /// The gossip scouting configuration.
531            pub gossip: #[derive(Default)]
532            GossipConf {
533                /// Whether gossip scouting is enabled or not.
534                enabled: Option<bool>,
535                /// When true, gossip scouting information are propagated multiple hops to all nodes in the local network.
536                /// When false, gossip scouting information are only propagated to the next hop.
537                /// Activating multihop gossip implies more scouting traffic and a lower scalability.
538                /// It mostly makes sense when using "linkstate" routing mode where all nodes in the subsystem don't have
539                /// direct connectivity with each other.
540                multihop: Option<bool>,
541                /// Which type of Zenoh instances to send gossip messages to.
542                target: Option<ModeDependentValue<WhatAmIMatcher>>,
543                /// Which type of Zenoh instances to automatically establish sessions with upon discovery through gossip.
544                autoconnect: Option<ModeDependentValue<WhatAmIMatcher>>,
545                /// Strategy for autoconnection, mainly to avoid nodes connecting to each other redundantly.
546                autoconnect_strategy: Option<ModeDependentValue<TargetDependentValue<AutoConnectStrategy>>>,
547            },
548        },
549
550        /// Configuration of data messages timestamps management.
551        pub timestamping: #[derive(Default)]
552        TimestampingConf {
553            /// Whether data messages should be timestamped if not already.
554            enabled: Option<ModeDependentValue<bool>>,
555            /// Whether data messages with timestamps in the future should be dropped or not.
556            /// If set to false (default), messages with timestamps in the future are retimestamped.
557            /// Timestamps are ignored if timestamping is disabled.
558            drop_future_timestamp: Option<bool>,
559        },
560
561        /// The default timeout to apply to queries in milliseconds.
562        queries_default_timeout: Option<u64>,
563
564        /// The routing strategy to use and it's configuration.
565        pub routing: #[derive(Default)]
566        RoutingConf {
567            /// The routing strategy to use in routers and it's configuration.
568            pub router: #[derive(Default)]
569            RouterRoutingConf {
570                /// When set to true a router will forward data between two peers
571                /// directly connected to it if it detects that those peers are not
572                /// connected to each other.
573                /// The failover brokering only works if gossip discovery is enabled.
574                peers_failover_brokering: Option<bool>,
575                /// Linkstate mode configuration.
576                pub linkstate: #[derive(Default)]
577                LinkstateConf {
578                    /// Weights of the outgoing links in linkstate mode.
579                    /// If none of the two endpoint nodes of a transport specifies its weight, a weight of 100 is applied.
580                    /// If only one of the two endpoint nodes of a transport specifies its weight, the specified weight is applied.
581                    /// If both endpoint nodes of a transport specify its weight, the greater weight is applied.
582                    pub transport_weights: Vec<TransportWeight>,
583                },
584            },
585            /// The routing strategy to use in peers and it's configuration.
586            pub peer: #[derive(Default)]
587            PeerRoutingConf {
588                /// The routing strategy to use in peers. ("peer_to_peer" or "linkstate").
589                /// This option needs to be set to the same value in all peers and routers of the subsystem.
590                mode: Option<String>,
591                /// Linkstate mode configuration (only taken into account if mode == "linkstate").
592                pub linkstate: LinkstateConf,
593            },
594            /// The interests-based routing configuration.
595            /// This configuration applies regardless of the mode (router, peer or client).
596            pub interests: #[derive(Default)]
597            InterestsConf {
598                /// The timeout to wait for incoming interests declarations.
599                timeout: Option<u64>,
600            },
601        },
602
603        /// The declarations aggregation strategy.
604        pub aggregation: #[derive(Default)]
605        AggregationConf {
606            /// A list of key-expressions for which all included subscribers will be aggregated into.
607            subscribers: Vec<OwnedKeyExpr>,
608            /// A list of key-expressions for which all included publishers will be aggregated into.
609            publishers: Vec<OwnedKeyExpr>,
610        },
611
612        /// Overwrite QoS options for Zenoh messages by key expression (ignores Zenoh API QoS config)
613        pub qos: #[derive(Default)]
614        QoSConfig {
615            /// A list of QoS configurations for PUT and DELETE messages by key expressions
616            publication: PublisherQoSConfList,
617            /// Configuration of the qos overwrite interceptor rules
618            network: Vec<QosOverwriteItemConf>,
619        },
620
621        pub transport: #[derive(Default)]
622        TransportConf {
623            pub unicast: TransportUnicastConf {
624                /// Timeout in milliseconds when opening a link (default: 10000).
625                open_timeout: u64,
626                /// Timeout in milliseconds when accepting a link (default: 10000).
627                accept_timeout: u64,
628                /// Number of links that may stay pending during accept phase (default: 100).
629                accept_pending: usize,
630                /// Maximum number of unicast sessions (default: 1000)
631                max_sessions: usize,
632                /// Maximum number of unicast incoming links per transport session (default: 1)
633                /// If set to a value greater than 1, multiple outgoing links are also allowed;
634                /// otherwise, only one outgoing link is allowed.
635                /// Issue https://github.com/eclipse-zenoh/zenoh/issues/1533
636                max_links: usize,
637                /// Enables the LowLatency transport (default `false`).
638                /// This option does not make LowLatency transport mandatory, the actual implementation of transport
639                /// used will depend on Establish procedure and other party's settings
640                lowlatency: bool,
641                pub qos: QoSUnicastConf {
642                    /// Whether QoS is enabled or not.
643                    /// If set to `false`, the QoS will be disabled. (default `true`).
644                    enabled: bool
645                },
646                pub compression: CompressionUnicastConf {
647                    /// You must compile zenoh with "transport_compression" feature to be able to enable compression.
648                    /// When enabled is true, batches will be sent compressed. (default `false`).
649                    enabled: bool,
650                },
651            },
652            pub multicast: TransportMulticastConf {
653                /// Link join interval duration in milliseconds (default: 2500)
654                join_interval: Option<u64>,
655                /// Maximum number of multicast sessions (default: 1000)
656                max_sessions: Option<usize>,
657                pub qos: QoSMulticastConf {
658                    /// Whether QoS is enabled or not.
659                    /// If set to `false`, the QoS will be disabled. (default `false`).
660                    enabled: bool
661                },
662                pub compression: CompressionMulticastConf {
663                    /// You must compile zenoh with "transport_compression" feature to be able to enable compression.
664                    /// When enabled is true, batches will be sent compressed. (default `false`).
665                    enabled: bool,
666                },
667            },
668            pub link: #[derive(Default)]
669            TransportLinkConf {
670                // An optional whitelist of protocols to be used for accepting and opening sessions.
671                // If not configured, all the supported protocols are automatically whitelisted.
672                pub protocols: Option<Vec<String>>,
673                pub tx: LinkTxConf {
674                    /// The resolution in bits to be used for the message sequence numbers.
675                    /// When establishing a session with another Zenoh instance, the lowest value of the two instances will be used.
676                    /// Accepted values: 8bit, 16bit, 32bit, 64bit.
677                    sequence_number_resolution: Bits where (sequence_number_resolution_validator),
678                    /// Link lease duration in milliseconds (default: 10000)
679                    lease: u64,
680                    /// Number of keep-alive messages in a link lease duration (default: 4)
681                    keep_alive: usize,
682                    /// Zenoh's MTU equivalent (default: 2^16-1) (max: 2^16-1)
683                    batch_size: BatchSize,
684                    pub queue: #[derive(Default)]
685                    QueueConf {
686                        /// The size of each priority queue indicates the number of batches a given queue can contain.
687                        /// The amount of memory being allocated for each queue is then SIZE_XXX * BATCH_SIZE.
688                        /// In the case of the transport link MTU being smaller than the ZN_BATCH_SIZE,
689                        /// then amount of memory being allocated for each queue is SIZE_XXX * LINK_MTU.
690                        /// If qos is false, then only the DATA priority will be allocated.
691                        pub size: QueueSizeConf {
692                            control: usize,
693                            real_time: usize,
694                            interactive_high: usize,
695                            interactive_low: usize,
696                            data_high: usize,
697                            data: usize,
698                            data_low: usize,
699                            background: usize,
700                        } where (queue_size_validator),
701                        /// Congestion occurs when the queue is empty (no available batch).
702                        /// Using CongestionControl::Block the caller is blocked until a batch is available and re-inserted into the queue.
703                        /// Using CongestionControl::Drop the message might be dropped, depending on conditions configured here.
704                        pub congestion_control: #[derive(Default)]
705                        CongestionControlConf {
706                            /// Behavior pushing CongestionControl::Drop messages to the queue.
707                            pub drop: CongestionControlDropConf {
708                                /// The maximum time in microseconds to wait for an available batch before dropping a droppable message
709                                /// if still no batch is available.
710                                wait_before_drop: i64,
711                                /// The maximum deadline limit for multi-fragment messages.
712                                max_wait_before_drop_fragments: i64,
713                            },
714                            /// Behavior pushing CongestionControl::Block messages to the queue.
715                            pub block: CongestionControlBlockConf {
716                                /// The maximum time in microseconds to wait for an available batch before closing the transport session
717                                /// when sending a blocking message if still no batch is available.
718                                wait_before_close: i64,
719                            },
720                        },
721                        pub batching: BatchingConf {
722                            /// Perform adaptive batching of messages if they are smaller of the batch_size.
723                            /// When the network is detected to not be fast enough to transmit every message individually, many small messages may be
724                            /// batched together and sent all at once on the wire reducing the overall network overhead. This is typically of a high-throughput
725                            /// scenario mainly composed of small messages. In other words, batching is activated by the network back-pressure.
726                            enabled: bool,
727                            /// The maximum time limit (in ms) a message should be retained for batching when back-pressure happens.
728                            time_limit: u64,
729                        },
730                        /// Perform lazy memory allocation of batches in the prioritiey queues. If set to false all batches are initialized at
731                        /// initialization time. If set to true the batches will be allocated when needed up to the maximum number of batches
732                        /// configured in the size configuration parameter.
733                        pub allocation: #[derive(Default, Copy, PartialEq, Eq)]
734                        QueueAllocConf {
735                            pub mode: QueueAllocMode,
736                        },
737                    },
738                    // Number of threads used for TX
739                    threads: usize,
740                },
741                pub rx: LinkRxConf {
742                    /// Receiving buffer size in bytes for each link
743                    /// The default the rx_buffer_size value is the same as the default batch size: 65535.
744                    /// For very high throughput scenarios, the rx_buffer_size can be increased to accommodate
745                    /// more in-flight data. This is particularly relevant when dealing with large messages.
746                    /// E.g. for 16MiB rx_buffer_size set the value to: 16777216.
747                    buffer_size: usize,
748                    /// Maximum size of the defragmentation buffer at receiver end (default: 1GiB).
749                    /// Fragmented messages that are larger than the configured size will be dropped.
750                    max_message_size: usize,
751                },
752                pub tls: #[derive(Default)]
753                TLSConf {
754                    root_ca_certificate: Option<String>,
755                    listen_private_key: Option<String>,
756                    listen_certificate: Option<String>,
757                    enable_mtls: Option<bool>,
758                    connect_private_key: Option<String>,
759                    connect_certificate: Option<String>,
760                    verify_name_on_connect: Option<bool>,
761                    close_link_on_expiration: Option<bool>,
762                    /// Configure TCP write buffer size
763                    pub so_sndbuf: Option<u32>,
764                    /// Configure TCP read buffer size
765                    pub so_rcvbuf: Option<u32>,
766                    // Skip serializing field because they contain secrets
767                    #[serde(skip_serializing)]
768                    root_ca_certificate_base64: Option<SecretValue>,
769                    #[serde(skip_serializing)]
770                    listen_private_key_base64:  Option<SecretValue>,
771                    #[serde(skip_serializing)]
772                    listen_certificate_base64: Option<SecretValue>,
773                    #[serde(skip_serializing)]
774                    connect_private_key_base64 :  Option<SecretValue>,
775                    #[serde(skip_serializing)]
776                    connect_certificate_base64 :  Option<SecretValue>,
777                },
778                pub tcp: #[derive(Default)]
779                TcpConf {
780                    /// Configure TCP write buffer size
781                    pub so_sndbuf: Option<u32>,
782                    /// Configure TCP read buffer size
783                    pub so_rcvbuf: Option<u32>,
784                },
785                pub unixpipe: #[derive(Default)]
786                UnixPipeConf {
787                    file_access_mask: Option<u32>
788                },
789            },
790            pub shared_memory:
791            ShmConf {
792                /// Whether shared memory is enabled or not.
793                /// If set to `true`, the SHM buffer optimization support will be announced to other parties. (default `true`).
794                /// This option doesn't make SHM buffer optimization mandatory, the real support depends on other party setting
795                /// A probing procedure for shared memory is performed upon session opening. To enable zenoh to operate
796                /// over shared memory (and to not fallback on network mode), shared memory needs to be enabled also on the
797                /// subscriber side. By doing so, the probing procedure will succeed and shared memory will operate as expected.
798                enabled: bool,
799                /// SHM resources initialization mode (default "lazy").
800                /// - "lazy": SHM subsystem internals will be initialized lazily upon the first SHM buffer
801                /// allocation or reception. This setting provides better startup time and optimizes resource usage,
802                /// but produces extra latency at the first SHM buffer interaction.
803                /// - "init": SHM subsystem internals will be initialized upon Session opening. This setting sacrifices
804                /// startup time, but guarantees no latency impact when first SHM buffer is processed.
805                mode: ShmInitMode,
806                pub transport_optimization:
807                LargeMessageTransportOpt {
808                    /// Enables transport optimization for large messages (default `true`).
809                    /// Implicitly puts large messages into shared memory for transports with SHM-compatible connection.
810                    enabled: bool,
811                    /// SHM arena size in bytes used for transport optimization (default `16 * 1024 * 1024`).
812                    pool_size: NonZeroUsize,
813                    /// Allow optimization for messages equal or larger than this threshold in bytes (default `3072`).
814                    message_size_threshold: usize,
815                },
816            },
817            pub auth: #[derive(Default)]
818            AuthConf {
819                /// The configuration of authentication.
820                /// A password implies a username is required.
821                pub usrpwd: #[derive(Default)]
822                UsrPwdConf {
823                    user: Option<String>,
824                    password: Option<String>,
825                    /// The path to a file containing the user password dictionary, a file containing `<user>:<password>`
826                    dictionary_file: Option<String>,
827                } where (user_conf_validator),
828                pub pubkey: #[derive(Default)]
829                PubKeyConf {
830                    public_key_pem: Option<String>,
831                    private_key_pem: Option<String>,
832                    public_key_file: Option<String>,
833                    private_key_file: Option<String>,
834                    key_size: Option<usize>,
835                    known_keys_file: Option<String>,
836                },
837            },
838
839        },
840        /// Configuration of the admin space.
841        pub adminspace: #[derive(Default)]
842        /// <div class="stab unstable">
843        ///   <span class="emoji">🔬</span>
844        ///   This API has been marked as unstable: it works as advertised, but we may change it in a future release.
845        ///   To use it, you must enable zenoh's <code>unstable</code> feature flag.
846        /// </div>
847        AdminSpaceConf {
848            /// Enable the admin space
849            #[serde(default = "set_false")]
850            pub enabled: bool,
851            /// Permissions on the admin space
852            pub permissions:
853            PermissionsConf {
854                /// Whether the admin space replies to queries (true by default).
855                #[serde(default = "set_true")]
856                pub read: bool,
857                /// Whether the admin space accepts config changes at runtime (false by default).
858                #[serde(default = "set_false")]
859                pub write: bool,
860            },
861
862        },
863
864        /// Namespace prefix.
865        /// If not None, all outgoing key expressions will be
866        /// automatically prefixed with specified string,
867        /// and all incoming key expressions will be stripped
868        /// of specified prefix.
869        /// Namespace is applied to the session.
870        /// E. g. if session has a namespace of "1" then session.put("my/keyexpr", message),
871        /// will put a message into "1/my/keyexpr". Same applies to all other operations within this session.
872        pub namespace: Option<OwnedNonWildKeyExpr>,
873
874        /// Configuration of the downsampling.
875        downsampling: Vec<DownsamplingItemConf> where (downsampling_validator),
876
877        /// Configuration of the access control (ACL)
878        pub access_control: AclConfig {
879            pub enabled: bool,
880            pub default_permission: Permission,
881            pub rules: Option<Vec<AclConfigRule>>,
882            pub subjects: Option<Vec<AclConfigSubjects>>,
883            pub policies: Option<Vec<AclConfigPolicyEntry>>,
884        },
885
886        /// Configuration of the low-pass filter
887        pub low_pass_filter: Vec<LowPassFilterConf>,
888
889        /// A list of directories where plugins may be searched for if no `__path__` was specified for them.
890        /// The executable's current directory will be added to the search paths.
891        pub plugins_loading: #[derive(Default)]
892        PluginsLoading {
893            pub enabled: bool,
894            pub search_dirs: LibSearchDirs,
895        },
896        #[validated(recursive_accessors)]
897        /// The configuration for plugins.
898        ///
899        /// Please refer to [`PluginsConfig`]'s documentation for further details.
900        plugins: PluginsConfig,
901    }
902}
903
904#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
905#[serde(rename_all = "snake_case")]
906pub enum QueueAllocMode {
907    Init,
908    #[default]
909    Lazy,
910}
911
912#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
913#[serde(rename_all = "snake_case")]
914pub enum ShmInitMode {
915    Init,
916    #[default]
917    Lazy,
918}
919
920impl Default for PermissionsConf {
921    fn default() -> Self {
922        PermissionsConf {
923            read: true,
924            write: false,
925        }
926    }
927}
928
929fn set_true() -> bool {
930    true
931}
932fn set_false() -> bool {
933    false
934}
935
936#[test]
937fn config_deser() {
938    let config = Config::from_deserializer(
939        &mut json5::Deserializer::from_str(
940            r#"{
941        scouting: {
942          multicast: {
943            enabled: false,
944            autoconnect: ["peer", "router"]
945          }
946        }
947      }"#,
948        )
949        .unwrap(),
950    )
951    .unwrap();
952    assert_eq!(*config.scouting().multicast().enabled(), Some(false));
953    assert_eq!(
954        config.scouting().multicast().autoconnect().router(),
955        Some(&WhatAmIMatcher::empty().router().peer())
956    );
957    assert_eq!(
958        config.scouting().multicast().autoconnect().peer(),
959        Some(&WhatAmIMatcher::empty().router().peer())
960    );
961    assert_eq!(
962        config.scouting().multicast().autoconnect().client(),
963        Some(&WhatAmIMatcher::empty().router().peer())
964    );
965    let config = Config::from_deserializer(
966        &mut json5::Deserializer::from_str(
967            r#"{
968        scouting: {
969          multicast: {
970            enabled: false,
971            autoconnect: {router: [], peer: ["peer", "router"]}
972          }
973        }
974      }"#,
975        )
976        .unwrap(),
977    )
978    .unwrap();
979    assert_eq!(*config.scouting().multicast().enabled(), Some(false));
980    assert_eq!(
981        config.scouting().multicast().autoconnect().router(),
982        Some(&WhatAmIMatcher::empty())
983    );
984    assert_eq!(
985        config.scouting().multicast().autoconnect().peer(),
986        Some(&WhatAmIMatcher::empty().router().peer())
987    );
988    assert_eq!(config.scouting().multicast().autoconnect().client(), None);
989    let config = Config::from_deserializer(
990        &mut json5::Deserializer::from_str(
991            r#"{transport: { auth: { usrpwd: { user: null, password: null, dictionary_file: "file" }}}}"#,
992        )
993            .unwrap(),
994    )
995        .unwrap();
996    assert_eq!(
997        config
998            .transport()
999            .auth()
1000            .usrpwd()
1001            .dictionary_file()
1002            .as_ref()
1003            .map(|s| s.as_ref()),
1004        Some("file")
1005    );
1006    std::mem::drop(Config::from_deserializer(
1007        &mut json5::Deserializer::from_str(
1008            r#"{transport: { auth: { usrpwd: { user: null, password: null, user_password_dictionary: "file" }}}}"#,
1009        )
1010            .unwrap(),
1011    )
1012        .unwrap_err());
1013
1014    let config = Config::from_deserializer(
1015        &mut json5::Deserializer::from_str(
1016            r#"{
1017              qos: {
1018                network: [
1019                  {
1020                    messages: ["put"],
1021                    overwrite: {
1022                      priority: "foo",
1023                    },
1024                  },
1025                ],
1026              }
1027            }"#,
1028        )
1029        .unwrap(),
1030    );
1031    assert!(config.is_err());
1032
1033    let config = Config::from_deserializer(
1034        &mut json5::Deserializer::from_str(
1035            r#"{
1036              qos: {
1037                network: [
1038                  {
1039                    messages: ["put"],
1040                    overwrite: {
1041                      priority: +8,
1042                    },
1043                  },
1044                ],
1045              }
1046            }"#,
1047        )
1048        .unwrap(),
1049    );
1050    assert!(config.is_err());
1051
1052    let config = Config::from_deserializer(
1053        &mut json5::Deserializer::from_str(
1054            r#"{
1055              qos: {
1056                network: [
1057                  {
1058                    messages: ["put"],
1059                    overwrite: {
1060                      priority: "data_high",
1061                    },
1062                  },
1063                ],
1064              }
1065            }"#,
1066        )
1067        .unwrap(),
1068    )
1069    .unwrap();
1070    assert_eq!(
1071        config.qos().network().first().unwrap().overwrite.priority,
1072        Some(qos::PriorityUpdateConf::Priority(
1073            qos::PriorityConf::DataHigh
1074        ))
1075    );
1076
1077    let config = Config::from_deserializer(
1078        &mut json5::Deserializer::from_str(
1079            r#"{
1080              qos: {
1081                network: [
1082                  {
1083                    messages: ["put"],
1084                    overwrite: {
1085                      priority: +1,
1086                    },
1087                  },
1088                ],
1089              }
1090            }"#,
1091        )
1092        .unwrap(),
1093    )
1094    .unwrap();
1095    assert_eq!(
1096        config.qos().network().first().unwrap().overwrite.priority,
1097        Some(qos::PriorityUpdateConf::Increment(1))
1098    );
1099
1100    let config = Config::from_deserializer(
1101        &mut json5::Deserializer::from_str(
1102            r#"{
1103              qos: {
1104                network: [
1105                  {
1106                    messages: ["put"],
1107                    payload_size: "0..99",
1108                    overwrite: {},
1109                  },
1110                ],
1111              }
1112            }"#,
1113        )
1114        .unwrap(),
1115    )
1116    .unwrap();
1117    assert_eq!(
1118        config
1119            .qos()
1120            .network()
1121            .first()
1122            .unwrap()
1123            .payload_size
1124            .as_ref()
1125            .map(|r| (r.start_bound(), r.end_bound())),
1126        Some((Bound::Included(&0), Bound::Included(&99)))
1127    );
1128
1129    let config = Config::from_deserializer(
1130        &mut json5::Deserializer::from_str(
1131            r#"{
1132              qos: {
1133                network: [
1134                  {
1135                    messages: ["put"],
1136                    payload_size: "100..",
1137                    overwrite: {},
1138                  },
1139                ],
1140              }
1141            }"#,
1142        )
1143        .unwrap(),
1144    )
1145    .unwrap();
1146    assert_eq!(
1147        config
1148            .qos()
1149            .network()
1150            .first()
1151            .unwrap()
1152            .payload_size
1153            .as_ref()
1154            .map(|r| (r.start_bound(), r.end_bound())),
1155        Some((Bound::Included(&100), Bound::Unbounded))
1156    );
1157
1158    let config = Config::from_deserializer(
1159        &mut json5::Deserializer::from_str(
1160            r#"{
1161              qos: {
1162                network: [
1163                  {
1164                    messages: ["put"],
1165                    qos: {
1166                      congestion_control: "drop",
1167                      priority: "data",
1168                      express: true,
1169                      reliability: "reliable",
1170                    },
1171                    overwrite: {},
1172                  },
1173                ],
1174              }
1175            }"#,
1176        )
1177        .unwrap(),
1178    )
1179    .unwrap();
1180    assert_eq!(
1181        config.qos().network().first().unwrap().qos,
1182        Some(QosFilter {
1183            congestion_control: Some(qos::CongestionControlConf::Drop),
1184            priority: Some(qos::PriorityConf::Data),
1185            express: Some(true),
1186            reliability: Some(qos::ReliabilityConf::Reliable),
1187        })
1188    );
1189
1190    dbg!(Config::from_file("../../DEFAULT_CONFIG.json5").unwrap());
1191}
1192
1193impl Config {
1194    pub fn insert<'d, D: serde::Deserializer<'d>>(
1195        &mut self,
1196        key: &str,
1197        value: D,
1198    ) -> Result<(), validated_struct::InsertionError>
1199    where
1200        validated_struct::InsertionError: From<D::Error>,
1201    {
1202        <Self as ValidatedMap>::insert(self, key, value)
1203    }
1204
1205    pub fn get(
1206        &self,
1207        key: &str,
1208    ) -> Result<<Self as ValidatedMapAssociatedTypes<'_>>::Accessor, GetError> {
1209        <Self as ValidatedMap>::get(self, key)
1210    }
1211
1212    pub fn get_json(&self, key: &str) -> Result<String, GetError> {
1213        <Self as ValidatedMap>::get_json(self, key)
1214    }
1215
1216    pub fn insert_json5(
1217        &mut self,
1218        key: &str,
1219        value: &str,
1220    ) -> Result<(), validated_struct::InsertionError> {
1221        <Self as ValidatedMap>::insert_json5(self, key, value)
1222    }
1223
1224    pub fn keys(&self) -> impl Iterator<Item = String> {
1225        <Self as ValidatedMap>::keys(self).into_iter()
1226    }
1227
1228    pub fn set_plugin_validator<T: ConfigValidator + 'static>(&mut self, validator: Weak<T>) {
1229        self.plugins.validator = validator;
1230    }
1231
1232    pub fn plugin(&self, name: &str) -> Option<&Value> {
1233        self.plugins.values.get(name)
1234    }
1235
1236    pub fn sift_privates(&self) -> Self {
1237        let mut copy = self.clone();
1238        copy.plugins.sift_privates();
1239        copy
1240    }
1241
1242    pub fn remove<K: AsRef<str>>(&mut self, key: K) -> ZResult<()> {
1243        let key = key.as_ref();
1244
1245        let key = key.strip_prefix('/').unwrap_or(key);
1246        if !key.starts_with("plugins/") {
1247            bail!(
1248                "Removal of values from Config is only supported for keys starting with `plugins/`"
1249            )
1250        }
1251        self.plugins.remove(&key["plugins/".len()..])
1252    }
1253
1254    pub fn get_retry_config(
1255        &self,
1256        endpoint: Option<&EndPoint>,
1257        listen: bool,
1258    ) -> ConnectionRetryConf {
1259        get_retry_config(self, endpoint, listen)
1260    }
1261}
1262
1263#[derive(Debug)]
1264pub enum ConfigOpenErr {
1265    IoError(std::io::Error),
1266    JsonParseErr(json5::Error),
1267    InvalidConfiguration(Box<Config>),
1268}
1269impl std::fmt::Display for ConfigOpenErr {
1270    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1271        match self {
1272            ConfigOpenErr::IoError(e) => write!(f, "Couldn't open file: {e}"),
1273            ConfigOpenErr::JsonParseErr(e) => write!(f, "JSON5 parsing error {e}"),
1274            ConfigOpenErr::InvalidConfiguration(c) => write!(
1275                f,
1276                "Invalid configuration {}",
1277                serde_json::to_string(c).unwrap()
1278            ),
1279        }
1280    }
1281}
1282impl std::error::Error for ConfigOpenErr {}
1283impl Config {
1284    pub fn from_file<P: AsRef<Path>>(path: P) -> ZResult<Self> {
1285        let path = path.as_ref();
1286        let mut config = Self::_from_file(path)?;
1287        config.plugins.load_external_configs()?;
1288        Ok(config)
1289    }
1290
1291    fn _from_file(path: &Path) -> ZResult<Config> {
1292        match std::fs::File::open(path) {
1293            Ok(mut f) => {
1294                let mut content = String::new();
1295                if let Err(e) = f.read_to_string(&mut content) {
1296                    bail!(e)
1297                }
1298                if content.is_empty() {
1299                    bail!("Empty config file");
1300                }
1301                match path
1302                    .extension()
1303                    .map(|s| s.to_str().unwrap())
1304                {
1305                    Some("json") | Some("json5") => match json5::Deserializer::from_str(&content) {
1306                        Ok(mut d) => Config::from_deserializer(&mut d).map_err(|e| match e {
1307                            Ok(c) => zerror!("Invalid configuration: {}", c).into(),
1308                            Err(e) => zerror!("JSON error: {:?}", e).into(),
1309                        }),
1310                        Err(e) => bail!(e),
1311                    },
1312                    Some("yaml") | Some("yml") => Config::from_deserializer(serde_yaml::Deserializer::from_str(&content)).map_err(|e| match e {
1313                        Ok(c) => zerror!("Invalid configuration: {}", c).into(),
1314                        Err(e) => zerror!("YAML error: {:?}", e).into(),
1315                    }),
1316                    Some(other) => bail!("Unsupported file type '.{}' (.json, .json5 and .yaml are supported)", other),
1317                    None => bail!("Unsupported file type. Configuration files must have an extension (.json, .json5 and .yaml supported)")
1318                }
1319            }
1320            Err(e) => bail!(e),
1321        }
1322    }
1323
1324    pub fn libloader(&self) -> LibLoader {
1325        if self.plugins_loading.enabled {
1326            LibLoader::new(self.plugins_loading.search_dirs().clone())
1327        } else {
1328            LibLoader::empty()
1329        }
1330    }
1331}
1332
1333impl std::fmt::Display for Config {
1334    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1335        serde_json::to_value(self)
1336            .map(|mut json| {
1337                sift_privates(&mut json);
1338                write!(f, "{json}")
1339            })
1340            .map_err(|e| {
1341                _ = write!(f, "{e:?}");
1342                fmt::Error
1343            })?
1344    }
1345}
1346
1347#[test]
1348fn config_from_json() {
1349    let from_str = serde_json::Deserializer::from_str;
1350    let mut config = Config::from_deserializer(&mut from_str(r#"{}"#)).unwrap();
1351    config
1352        .insert("transport/link/tx/lease", &mut from_str("168"))
1353        .unwrap();
1354    dbg!(std::mem::size_of_val(&config));
1355    println!("{}", serde_json::to_string_pretty(&config).unwrap());
1356}
1357
1358fn sequence_number_resolution_validator(b: &Bits) -> bool {
1359    b <= &Bits::from(TransportSn::MAX)
1360}
1361
1362fn queue_size_validator(q: &QueueSizeConf) -> bool {
1363    fn check(size: &usize) -> bool {
1364        (QueueSizeConf::MIN..=QueueSizeConf::MAX).contains(size)
1365    }
1366
1367    let QueueSizeConf {
1368        control,
1369        real_time,
1370        interactive_low,
1371        interactive_high,
1372        data_high,
1373        data,
1374        data_low,
1375        background,
1376    } = q;
1377    check(control)
1378        && check(real_time)
1379        && check(interactive_low)
1380        && check(interactive_high)
1381        && check(data_high)
1382        && check(data)
1383        && check(data_low)
1384        && check(background)
1385}
1386
1387fn user_conf_validator(u: &UsrPwdConf) -> bool {
1388    (u.password().is_none() && u.user().is_none()) || (u.password().is_some() && u.user().is_some())
1389}
1390
1391/// This part of the configuration is highly dynamic (any [`serde_json::Value`] may be put in there), but should follow this scheme:
1392/// ```javascript
1393/// plugins: {
1394///     // `plugin_name` must be unique per configuration, and will be used to find the appropriate
1395///     // dynamic library to load if no `__path__` is specified
1396///     [plugin_name]: {
1397///         // Defaults to `false`. Setting this to `true` does 2 things:
1398///         // * If `zenohd` fails to locate the requested plugin, it will crash instead of logging an error.
1399///         // * Plugins are expected to check this value to set their panic-behaviour: plugins are encouraged
1400///         //   to panic upon non-recoverable errors if their `__required__` flag is set to `true`, and to
1401///         //   simply log them otherwise
1402///         __required__: bool,
1403///         // The path(s) where the plugin is expected to be located.
1404///         // If none is specified, `zenohd` will search for a `<dylib_prefix>zenoh_plugin_<plugin_name>.<dylib_suffix>` file in the search directories.
1405///         // If any path is specified, file-search will be disabled, and the first path leading to
1406///         // an existing file will be used
1407///         __path__: string | [string],
1408///         // [plugin_name] may require additional configuration
1409///         ...
1410///     }
1411/// }
1412/// ```
1413#[derive(Clone)]
1414pub struct PluginsConfig {
1415    values: Value,
1416    validator: std::sync::Weak<dyn ConfigValidator>,
1417}
1418fn sift_privates(value: &mut serde_json::Value) {
1419    match value {
1420        Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) => {}
1421        Value::Array(a) => a.iter_mut().for_each(sift_privates),
1422        Value::Object(o) => {
1423            o.remove("private");
1424            o.values_mut().for_each(sift_privates);
1425        }
1426    }
1427}
1428
1429fn load_external_plugin_config(title: &str, value: &mut Value) -> ZResult<()> {
1430    let Some(values) = value.as_object_mut() else {
1431        bail!("{} must be object", title);
1432    };
1433    recursive_include(title, values, HashSet::new(), "__config__", ".")
1434}
1435
1436#[derive(Debug, Clone)]
1437pub struct PluginLoad {
1438    pub id: String,
1439    pub name: String,
1440    pub paths: Option<Vec<String>>,
1441    pub required: bool,
1442}
1443impl PluginsConfig {
1444    pub fn sift_privates(&mut self) {
1445        sift_privates(&mut self.values);
1446    }
1447    fn load_external_configs(&mut self) -> ZResult<()> {
1448        let Some(values) = self.values.as_object_mut() else {
1449            bail!("plugins configuration must be an object")
1450        };
1451        for (name, value) in values.iter_mut() {
1452            load_external_plugin_config(format!("plugins.{}", name.as_str()).as_str(), value)?;
1453        }
1454        Ok(())
1455    }
1456    pub fn load_requests(&'_ self) -> impl Iterator<Item = PluginLoad> + '_ {
1457        self.values.as_object().unwrap().iter().map(|(id, value)| {
1458            let value = value.as_object().expect("Plugin configurations must be objects");
1459            let required = match value.get("__required__") {
1460                None => false,
1461                Some(Value::Bool(b)) => *b,
1462                _ => panic!("Plugin '{id}' has an invalid '__required__' configuration property (must be a boolean)")
1463            };
1464            let name = match value.get("__plugin__") {
1465                Some(Value::String(p)) => p,
1466                _ => id,
1467            };
1468
1469            if let Some(paths) = value.get("__path__") {
1470                let paths = match paths {
1471                    Value::String(s) => vec![s.clone()],
1472                    Value::Array(a) => a.iter().map(|s| if let Value::String(s) = s { s.clone() } else { panic!("Plugin '{id}' has an invalid '__path__' configuration property (must be either string or array of strings)") }).collect(),
1473                    _ => panic!("Plugin '{id}' has an invalid '__path__' configuration property (must be either string or array of strings)")
1474                };
1475                PluginLoad { id: id.clone(), name: name.clone(), paths: Some(paths), required }
1476            } else {
1477                PluginLoad { id: id.clone(), name: name.clone(), paths: None, required }
1478            }
1479        })
1480    }
1481    pub fn remove(&mut self, key: &str) -> ZResult<()> {
1482        let mut split = key.split('/');
1483        let plugin = split.next().unwrap();
1484        let mut current = match split.next() {
1485            Some(first_in_plugin) => first_in_plugin,
1486            None => {
1487                self.values.as_object_mut().unwrap().remove(plugin);
1488                return Ok(());
1489            }
1490        };
1491        let (old_conf, mut new_conf) = match self.values.get_mut(plugin) {
1492            Some(plugin) => {
1493                let clone = plugin.clone();
1494                (plugin, clone)
1495            }
1496            None => bail!("No plugin {} to edit", plugin),
1497        };
1498        let mut remove_from = &mut new_conf;
1499        for next in split {
1500            match remove_from {
1501                Value::Object(o) => match o.get_mut(current) {
1502                    Some(v) => {
1503                        remove_from = unsafe {
1504                            std::mem::transmute::<&mut serde_json::Value, &mut serde_json::Value>(v)
1505                        }
1506                    }
1507                    None => bail!("{:?} has no {} property", o, current),
1508                },
1509                Value::Array(a) => {
1510                    let index: usize = current.parse()?;
1511                    if a.len() <= index {
1512                        bail!("{:?} cannot be indexed at {}", a, index)
1513                    }
1514                    remove_from = &mut a[index];
1515                }
1516                other => bail!("{} cannot be indexed", other),
1517            }
1518            current = next
1519        }
1520        match remove_from {
1521            Value::Object(o) => {
1522                if o.remove(current).is_none() {
1523                    bail!("{:?} has no {} property", o, current)
1524                }
1525            }
1526            Value::Array(a) => {
1527                let index: usize = current.parse()?;
1528                if a.len() <= index {
1529                    bail!("{:?} cannot be indexed at {}", a, index)
1530                }
1531                a.remove(index);
1532            }
1533            other => bail!("{} cannot be indexed", other),
1534        }
1535        let new_conf = if let Some(validator) = self.validator.upgrade() {
1536            match validator.check_config(
1537                plugin,
1538                &key[("plugins/".len() + plugin.len())..],
1539                old_conf.as_object().unwrap(),
1540                new_conf.as_object().unwrap(),
1541            )? {
1542                None => new_conf,
1543                Some(new_conf) => Value::Object(new_conf),
1544            }
1545        } else {
1546            new_conf
1547        };
1548        *old_conf = new_conf;
1549        Ok(())
1550    }
1551}
1552impl serde::Serialize for PluginsConfig {
1553    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
1554    where
1555        S: serde::Serializer,
1556    {
1557        let mut value = self.values.clone();
1558        sift_privates(&mut value);
1559        value.serialize(serializer)
1560    }
1561}
1562impl Default for PluginsConfig {
1563    fn default() -> Self {
1564        Self {
1565            values: Value::Object(Default::default()),
1566            validator: std::sync::Weak::<()>::new(),
1567        }
1568    }
1569}
1570impl<'a> serde::Deserialize<'a> for PluginsConfig {
1571    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
1572    where
1573        D: serde::Deserializer<'a>,
1574    {
1575        Ok(PluginsConfig {
1576            values: serde::Deserialize::deserialize(deserializer)?,
1577            validator: std::sync::Weak::<()>::new(),
1578        })
1579    }
1580}
1581
1582impl std::fmt::Debug for PluginsConfig {
1583    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1584        let mut values: Value = self.values.clone();
1585        sift_privates(&mut values);
1586        write!(f, "{values:?}")
1587    }
1588}
1589
1590trait PartialMerge: Sized {
1591    fn merge(self, path: &str, value: Self) -> Result<Self, validated_struct::InsertionError>;
1592}
1593impl PartialMerge for serde_json::Value {
1594    fn merge(
1595        mut self,
1596        path: &str,
1597        new_value: Self,
1598    ) -> Result<Self, validated_struct::InsertionError> {
1599        let mut value = &mut self;
1600        let mut key = path;
1601        let key_not_found = || {
1602            Err(validated_struct::InsertionError::String(format!(
1603                "{path} not found"
1604            )))
1605        };
1606        while !key.is_empty() {
1607            let (current, new_key) = validated_struct::split_once(key, '/');
1608            key = new_key;
1609            if current.is_empty() {
1610                continue;
1611            }
1612            value = match value {
1613                Value::Bool(_) | Value::Number(_) | Value::String(_) => return key_not_found(),
1614                Value::Null => match current {
1615                    "0" | "+" => {
1616                        *value = Value::Array(vec![Value::Null]);
1617                        &mut value[0]
1618                    }
1619                    _ => {
1620                        *value = Value::Object(Default::default());
1621                        value
1622                            .as_object_mut()
1623                            .unwrap()
1624                            .entry(current)
1625                            .or_insert(Value::Null)
1626                    }
1627                },
1628                Value::Array(a) => match current {
1629                    "+" => {
1630                        a.push(Value::Null);
1631                        a.last_mut().unwrap()
1632                    }
1633                    "0" if a.is_empty() => {
1634                        a.push(Value::Null);
1635                        a.last_mut().unwrap()
1636                    }
1637                    _ => match current.parse::<usize>() {
1638                        Ok(i) => match a.get_mut(i) {
1639                            Some(r) => r,
1640                            None => return key_not_found(),
1641                        },
1642                        Err(_) => return key_not_found(),
1643                    },
1644                },
1645                Value::Object(v) => v.entry(current).or_insert(Value::Null),
1646            }
1647        }
1648        *value = new_value;
1649        Ok(self)
1650    }
1651}
1652impl<'a> validated_struct::ValidatedMapAssociatedTypes<'a> for PluginsConfig {
1653    type Accessor = &'a dyn Any;
1654}
1655impl validated_struct::ValidatedMap for PluginsConfig {
1656    fn insert<'d, D: serde::Deserializer<'d>>(
1657        &mut self,
1658        key: &str,
1659        deserializer: D,
1660    ) -> Result<(), validated_struct::InsertionError>
1661    where
1662        validated_struct::InsertionError: From<D::Error>,
1663    {
1664        let (plugin, key) = validated_struct::split_once(key, '/');
1665        let new_value: Value = serde::Deserialize::deserialize(deserializer)?;
1666        let value = self
1667            .values
1668            .as_object_mut()
1669            .unwrap()
1670            .entry(plugin)
1671            .or_insert(Value::Null);
1672        let new_value = value.clone().merge(key, new_value)?;
1673        *value = if let Some(validator) = self.validator.upgrade() {
1674            // New plugin configuration for compare with original configuration.
1675            // Return error if it's not an object.
1676            // Note: it's ok if original "new_value" is not an object: this can be some subkey of the plugin configuration. But the result of the merge should be an object.
1677            // Error occurs  if the original plugin configuration is not an object itself (e.g. null).
1678            let Some(new_plugin_config) = new_value.as_object() else {
1679                return Err(format!(
1680                    "Attempt to provide non-object value as configuration for plugin `{plugin}`"
1681                )
1682                .into());
1683            };
1684            // Original plugin configuration for compare with new configuration.
1685            // If for some reason it's not defined or not an object, we default to an empty object.
1686            // Usually this happens when no plugin with this name defined. Reject then should be performed by the validator with `plugin not found` error.
1687            let empty_config = Map::new();
1688            let current_plugin_config = value.as_object().unwrap_or(&empty_config);
1689            match validator.check_config(plugin, key, current_plugin_config, new_plugin_config) {
1690                // Validator made changes to the proposed configuration, take these changes
1691                Ok(Some(val)) => Value::Object(val),
1692                // Validator accepted the proposed configuration as is
1693                Ok(None) => new_value,
1694                // Validator rejected the proposed configuration
1695                Err(e) => return Err(format!("{e}").into()),
1696            }
1697        } else {
1698            new_value
1699        };
1700        Ok(())
1701    }
1702    fn get<'a>(&'a self, mut key: &str) -> Result<&'a dyn Any, GetError> {
1703        let (current, new_key) = validated_struct::split_once(key, '/');
1704        key = new_key;
1705        let mut value = match self.values.get(current) {
1706            Some(matched) => matched,
1707            None => return Err(GetError::NoMatchingKey),
1708        };
1709        while !key.is_empty() {
1710            let (current, new_key) = validated_struct::split_once(key, '/');
1711            key = new_key;
1712            let matched = match value {
1713                serde_json::Value::Null
1714                | serde_json::Value::Bool(_)
1715                | serde_json::Value::Number(_)
1716                | serde_json::Value::String(_) => return Err(GetError::NoMatchingKey),
1717                serde_json::Value::Array(a) => a.get(match current.parse::<usize>() {
1718                    Ok(i) => i,
1719                    Err(_) => return Err(GetError::NoMatchingKey),
1720                }),
1721                serde_json::Value::Object(v) => v.get(current),
1722            };
1723            value = match matched {
1724                Some(matched) => matched,
1725                None => return Err(GetError::NoMatchingKey),
1726            }
1727        }
1728        Ok(value)
1729    }
1730
1731    type Keys = Vec<String>;
1732    fn keys(&self) -> Self::Keys {
1733        self.values.as_object().unwrap().keys().cloned().collect()
1734    }
1735
1736    fn get_json(&self, mut key: &str) -> Result<String, GetError> {
1737        let (current, new_key) = validated_struct::split_once(key, '/');
1738        key = new_key;
1739        let mut value = match self.values.get(current) {
1740            Some(matched) => matched,
1741            None => return Err(GetError::NoMatchingKey),
1742        };
1743        while !key.is_empty() {
1744            let (current, new_key) = validated_struct::split_once(key, '/');
1745            key = new_key;
1746            let matched = match value {
1747                serde_json::Value::Null
1748                | serde_json::Value::Bool(_)
1749                | serde_json::Value::Number(_)
1750                | serde_json::Value::String(_) => return Err(GetError::NoMatchingKey),
1751                serde_json::Value::Array(a) => a.get(match current.parse::<usize>() {
1752                    Ok(i) => i,
1753                    Err(_) => return Err(GetError::NoMatchingKey),
1754                }),
1755                serde_json::Value::Object(v) => v.get(current),
1756            };
1757            value = match matched {
1758                Some(matched) => matched,
1759                None => return Err(GetError::NoMatchingKey),
1760            }
1761        }
1762        Ok(serde_json::to_string(value).unwrap())
1763    }
1764}
1765
1766#[macro_export]
1767macro_rules! unwrap_or_default {
1768    ($val:ident$(.$field:ident($($param:ident)?))*) => {
1769        $val$(.$field($($param)?))*.clone().unwrap_or(zenoh_config::defaults$(::$field$(($param))?)*.into())
1770    };
1771}
1772
1773pub trait IConfig {
1774    fn get(&self, key: &str) -> ZResult<String>;
1775    fn queries_default_timeout_ms(&self) -> u64;
1776    fn insert_json5(&self, key: &str, value: &str) -> ZResult<()>;
1777}
1778
1779pub struct GenericConfig(Arc<dyn IConfig>);
1780
1781impl Deref for GenericConfig {
1782    type Target = Arc<dyn IConfig>;
1783
1784    fn deref(&self) -> &Self::Target {
1785        &self.0
1786    }
1787}
1788
1789impl GenericConfig {
1790    pub fn new(value: Arc<dyn IConfig>) -> Self {
1791        GenericConfig(value)
1792    }
1793
1794    pub fn get_typed<T: for<'a> Deserialize<'a>>(&self, key: &str) -> ZResult<T> {
1795        self.0
1796            .get(key)
1797            .and_then(|v| serde_json::from_str::<T>(&v).map_err(|e| e.into()))
1798    }
1799
1800    pub fn get_plugin_config(&self, plugin_name: &str) -> ZResult<Value> {
1801        self.get(&("plugins/".to_owned() + plugin_name))
1802            .and_then(|v| serde_json::from_str(&v).map_err(|e| e.into()))
1803    }
1804}
zenoh_config/lib.rs

zenoh_config/
lib.rs
