zenoh_protocol/zenoh/
query.rs

1//
2// Copyright (c) 2022 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//
14use alloc::{string::String, vec::Vec};
15
16use serde::Deserialize;
17
18use crate::common::ZExtUnknown;
19
20/// The kind of consolidation to apply to a query.
21#[repr(u8)]
22#[derive(Debug, Default, Clone, PartialEq, Eq, Copy, Deserialize)]
23pub enum ConsolidationMode {
24    /// Apply automatic consolidation based on queryable's preferences
25    #[default]
26    Auto,
27    /// No consolidation applied: multiple samples may be received for the same key-timestamp.
28    None,
29    /// Monotonic consolidation immediately forwards samples, except if one with an equal or more recent timestamp
30    /// has already been sent with the same key.
31    ///
32    /// This optimizes latency while potentially reducing bandwidth.
33    ///
34    /// Note that this doesn't cause re-ordering, but drops the samples for which a more recent timestamp has already
35    /// been observed with the same key.
36    Monotonic,
37    /// Holds back samples to only send the set of samples that had the highest timestamp for their key.    
38    Latest,
39    // Remove the duplicates of any samples based on the their timestamp.
40    // Unique,
41}
42
43impl ConsolidationMode {
44    pub const DEFAULT: Self = Self::Auto;
45
46    #[cfg(feature = "test")]
47    #[doc(hidden)]
48    pub fn rand() -> Self {
49        use rand::prelude::SliceRandom;
50        let mut rng = rand::thread_rng();
51
52        *[Self::None, Self::Monotonic, Self::Latest, Self::Auto]
53            .choose(&mut rng)
54            .unwrap()
55    }
56}
57
58/// # Query message
59///
60/// ```text
61/// Flags:
62/// - C: Consolidation  if C==1 then consolidation is present
63/// - P: Parameters     If P==1 then the parameters are present
64/// - Z: Extension      If Z==1 then at least one extension is present
65///
66///   7 6 5 4 3 2 1 0
67///  +-+-+-+-+-+-+-+-+
68///  |Z|P|C|  QUERY  |
69///  +-+-+-+---------+
70///  % consolidation %  if C==1
71///  +---------------+
72///  ~ ps: <u8;z16>  ~  if P==1
73///  +---------------+
74///  ~  [qry_exts]   ~  if Z==1
75///  +---------------+
76/// ```
77pub mod flag {
78    pub const C: u8 = 1 << 5; // 0x20 Consolidation if C==1 then consolidation is present
79    pub const P: u8 = 1 << 6; // 0x40 Parameters    if P==1 then the parameters are present
80    pub const Z: u8 = 1 << 7; // 0x80 Extensions    if Z==1 then an extension will follow
81}
82
83#[derive(Debug, Clone, PartialEq, Eq)]
84pub struct Query {
85    pub consolidation: ConsolidationMode,
86    pub parameters: String,
87    pub ext_sinfo: Option<ext::SourceInfoType>,
88    pub ext_body: Option<ext::QueryBodyType>,
89    pub ext_attachment: Option<ext::AttachmentType>,
90    pub ext_unknown: Vec<ZExtUnknown>,
91}
92
93pub mod ext {
94    use crate::{common::ZExtZBuf, zextzbuf};
95
96    /// # SourceInfo extension
97    /// Used to carry additional information about the source of data
98    pub type SourceInfo = zextzbuf!(0x1, false);
99    pub type SourceInfoType = crate::zenoh::ext::SourceInfoType<{ SourceInfo::ID }>;
100
101    /// # QueryBody extension
102    /// Used to carry a body attached to the query
103    /// Shared Memory extension is automatically defined by ValueType extension if
104    /// #[cfg(feature = "shared-memory")] is defined.
105    pub type QueryBodyType = crate::zenoh::ext::ValueType<{ ZExtZBuf::<0x03>::id(false) }, 0x04>;
106
107    /// # User attachment
108    pub type Attachment = zextzbuf!(0x5, false);
109    pub type AttachmentType = crate::zenoh::ext::AttachmentType<{ Attachment::ID }>;
110}
111
112impl Query {
113    #[cfg(feature = "test")]
114    #[doc(hidden)]
115    pub fn rand() -> Self {
116        use rand::{
117            distributions::{Alphanumeric, DistString},
118            Rng,
119        };
120
121        use crate::common::iext;
122        let mut rng = rand::thread_rng();
123
124        const MIN: usize = 2;
125        const MAX: usize = 16;
126
127        let consolidation = ConsolidationMode::rand();
128        let parameters: String = if rng.gen_bool(0.5) {
129            let len = rng.gen_range(MIN..MAX);
130            Alphanumeric.sample_string(&mut rng, len)
131        } else {
132            String::new()
133        };
134        let ext_sinfo = rng.gen_bool(0.5).then_some(ext::SourceInfoType::rand());
135        let ext_body = rng.gen_bool(0.5).then_some(ext::QueryBodyType::rand());
136        let ext_attachment = rng.gen_bool(0.5).then_some(ext::AttachmentType::rand());
137        let mut ext_unknown = Vec::new();
138        for _ in 0..rng.gen_range(0..4) {
139            ext_unknown.push(ZExtUnknown::rand2(
140                iext::mid(ext::Attachment::ID) + 1,
141                false,
142            ));
143        }
144
145        Self {
146            consolidation,
147            parameters,
148            ext_sinfo,
149            ext_body,
150            ext_attachment,
151            ext_unknown,
152        }
153    }
154}
zenoh_protocol/zenoh/query.rs

zenoh_protocol/zenoh/
query.rs
