zenoh_flow/model/descriptor/node/
mod.rs

1//
2// Copyright (c) 2021 - 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
15pub mod operator;
16
17pub use operator::{CompositeOperatorDescriptor, OperatorDescriptor};
18use std::path::PathBuf;
19pub mod sink;
20pub use sink::SinkDescriptor;
21pub mod source;
22pub use source::SourceDescriptor;
23
24use crate::model::descriptor::{LinkDescriptor, Vars};
25use crate::model::{Middleware, ZFUri};
26use crate::runtime::dataflow::instance::builtin::zenoh::{
27    get_zenoh_sink_descriptor, get_zenoh_source_descriptor,
28};
29use crate::types::configuration::Merge;
30use crate::types::{Configuration, NodeId};
31use crate::utils::parse_uri;
32use crate::zfresult::ErrorKind;
33use crate::{bail, zferror, Result};
34use serde::{Deserialize, Serialize};
35
36/// Describes an node of the graph
37///
38/// ```yaml
39/// id : PrintSink
40/// descriptor: file://./target/release/counter_source.yaml
41/// configuration:
42///   start: 10
43///
44#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
45pub struct NodeDescriptor {
46    pub id: NodeId,
47    pub descriptor: String,
48    pub configuration: Option<Configuration>,
49}
50
51impl std::fmt::Display for NodeDescriptor {
52    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
53        write!(
54            f,
55            "ID: {} Descriptor: {} Configuration: {:?}",
56            self.id, self.descriptor, self.configuration
57        )
58    }
59}
60
61impl NodeDescriptor {
62    /// Creates a new `NodeDescriptor` from its YAML representation.
63    ///
64    ///  # Errors
65    /// A variant error is returned if deserialization fails.
66    pub fn from_yaml(data: &str) -> Result<Self> {
67        let dataflow_descriptor = serde_yaml::from_str::<NodeDescriptor>(data)
68            .map_err(|e| zferror!(ErrorKind::ParsingError, e))?;
69        Ok(dataflow_descriptor)
70    }
71
72    /// Creates a new `NodeDescriptor` from its JSON representation.
73    ///
74    ///  # Errors
75    /// A variant error is returned if deserialization fails.
76    pub fn from_json(data: &str) -> Result<Self> {
77        let dataflow_descriptor = serde_json::from_str::<NodeDescriptor>(data)
78            .map_err(|e| zferror!(ErrorKind::ParsingError, e))?;
79        Ok(dataflow_descriptor)
80    }
81
82    /// Returns the JSON representation of the `NodeDescriptor`.
83    ///
84    ///  # Errors
85    /// A variant error is returned if serialization fails.
86    pub fn to_json(&self) -> Result<String> {
87        serde_json::to_string(&self).map_err(|e| zferror!(ErrorKind::SerializationError, e).into())
88    }
89
90    /// Returns the YAML representation of the `NodeDescriptor`.
91    ///
92    ///  # Errors
93    /// A variant error is returned if serialization fails.
94    pub fn to_yaml(&self) -> Result<String> {
95        serde_yaml::to_string(&self).map_err(|e| zferror!(ErrorKind::SerializationError, e).into())
96    }
97
98    /// Flattens the `NodeDescriptor` by loading all the composite operators
99    ///
100    /// # Errors
101    ///
102    /// A variant error is returned if loading operators fails. Or if the
103    /// node does not contains an operator
104    pub async fn flatten(
105        self,
106        id: NodeId,
107        links: &mut Vec<LinkDescriptor>,
108        global_configuration: Option<Configuration>,
109        ancestors: &mut Vec<String>,
110    ) -> Result<Vec<OperatorDescriptor>> {
111        log::trace!("[Descriptor] loading operator {}", self.id);
112        let descriptor = match parse_uri(&self.descriptor)? {
113            crate::model::ZFUri::File(path) => try_load_descriptor_from_file(path).await,
114            crate::model::ZFUri::Builtin(_) => bail!(
115                ErrorKind::ConfigurationError,
116                "Builtin operators are not yet supported!"
117            ),
118        }?;
119
120        // We try to load the descriptor, first we try as simple one, if it fails we try as a
121        // composite one, if that also fails it is malformed.
122        let res_simple = OperatorDescriptor::from_yaml(&descriptor);
123        if let Ok(mut simple_operator) = res_simple {
124            log::trace!("[Descriptor] Operator {} is simple", simple_operator.id);
125            simple_operator.configuration = global_configuration
126                .clone()
127                .merge_overwrite(simple_operator.configuration);
128            simple_operator.id = id;
129            return Ok(vec![simple_operator]);
130        }
131
132        let res_composite = CompositeOperatorDescriptor::from_yaml(&descriptor);
133        if let Ok(composite_operator) = res_composite {
134            log::trace!(
135                "[Descriptor] Operator {} is composite",
136                composite_operator.id
137            );
138            if let Ok(index) = ancestors.binary_search(&self.descriptor) {
139                log::error!(
140                    "Possible recursion detected, < {} > would be included again after: {:?}",
141                    self.descriptor,
142                    &ancestors[index..]
143                );
144                bail!(
145                    ErrorKind::GenericError, // FIXME Dedicated error?
146                    "Possible recursion detected, < {} > would be included again after: {:?}",
147                    self.descriptor,
148                    &ancestors[index..]
149                );
150            }
151
152            ancestors.push(self.descriptor.clone());
153            let res = composite_operator
154                .flatten(id, links, global_configuration, ancestors)
155                .await;
156            ancestors.pop();
157
158            return res;
159        }
160
161        log::error!("Could not parse operator < {} >", self.descriptor);
162        log::error!("(Operator) {:?}", res_simple.err().unwrap());
163        log::error!("(Composite) {:?}", res_composite.err().unwrap());
164
165        bail!(
166            ErrorKind::ParsingError,
167            "Could not parse operator < {} >",
168            self.descriptor
169        )
170    }
171
172    /// Loads the source from the `NodeDescriptor`
173    ///
174    ///  # Errors
175    /// A variant error is returned if loading source fails. Or if the
176    ///  node does not contains an source
177    pub async fn load_source(
178        self,
179        global_configuration: Option<Configuration>,
180    ) -> Result<SourceDescriptor> {
181        log::trace!("[Descriptor] Loading Source {}", self.id);
182
183        match parse_uri(&self.descriptor)? {
184            ZFUri::File(path) => {
185                let descriptor = try_load_descriptor_from_file(path).await?;
186
187                match SourceDescriptor::from_yaml(&descriptor) {
188                    Ok(mut desc) => {
189                        desc.id = self.id;
190                        desc.configuration =
191                            global_configuration.merge_overwrite(desc.configuration);
192                        Ok(desc)
193                    }
194                    Err(e) => {
195                        log::warn!("Unable to read descriptor {}, error {}", self.id, e);
196                        Err(e)
197                    }
198                }
199            }
200            ZFUri::Builtin(mw) => match mw {
201                Middleware::Zenoh => match &self.configuration {
202                    Some(configuration) => {
203                        let mut desc = get_zenoh_source_descriptor(configuration)?;
204                        desc.id = self.id;
205                        desc.configuration =
206                            global_configuration.merge_overwrite(desc.configuration);
207                        Ok(desc)
208                    }
209                    None => {
210                        bail!(
211                            ErrorKind::MissingConfiguration,
212                            "Builtin Zenoh Sink needs a configuration!"
213                        )
214                    }
215                },
216            },
217        }
218    }
219
220    /// Loads the sink from the `NodeDescriptor`
221    ///
222    /// # Errors
223    ///
224    /// A variant error is returned if loading sink fails. Or if the
225    /// node does not contains an sink
226    pub async fn load_sink(
227        self,
228        global_configuration: Option<Configuration>,
229    ) -> Result<SinkDescriptor> {
230        log::trace!("[Descriptor] Loading sink {}", self.id);
231
232        match parse_uri(&self.descriptor)? {
233            ZFUri::File(path) => {
234                let descriptor = try_load_descriptor_from_file(path).await?;
235
236                match SinkDescriptor::from_yaml(&descriptor) {
237                    Ok(mut desc) => {
238                        desc.id = self.id;
239                        desc.configuration =
240                            global_configuration.merge_overwrite(desc.configuration);
241                        Ok(desc)
242                    }
243                    Err(e) => {
244                        log::warn!("Unable to read descriptor {}, error {}", self.id, e);
245                        Err(e)
246                    }
247                }
248            }
249            ZFUri::Builtin(mw) => match mw {
250                Middleware::Zenoh => match &self.configuration {
251                    Some(configuration) => {
252                        let mut desc = get_zenoh_sink_descriptor(configuration)?;
253                        desc.id = self.id;
254                        desc.configuration =
255                            global_configuration.merge_overwrite(desc.configuration);
256                        Ok(desc)
257                    }
258                    None => {
259                        bail!(
260                            ErrorKind::MissingConfiguration,
261                            "Builtin Zenoh Sink needs a configuration!"
262                        )
263                    }
264                },
265            },
266        }
267    }
268}
269
270/// Attempt to asynchronously read the content of the file pointed at by the `descriptor_path`.
271///
272/// This function will also expand the mustache notations present (if there are any).
273///
274/// # Errors
275///
276/// This function will return an error in the following situations:
277/// - The provided `descriptor_path` is incorrect, i.e. the file does not exists.
278/// - The content of the file could not be read.
279async fn try_load_descriptor_from_file(descriptor_path: PathBuf) -> Result<String> {
280    let data = async_std::fs::read_to_string(&descriptor_path).await?;
281    Vars::expand_mustache_yaml(&data)
282}
zenoh_flow/model/descriptor/node/mod.rs

zenoh_flow/model/descriptor/node/
mod.rs
