zenoh_flow/runtime/dataflow/
node.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
15use crate::model::record::{OperatorRecord, SinkRecord, SourceRecord};
16use crate::prelude::{Configuration, Context, Inputs, Node, Outputs, Result};
17use std::ops::Deref;
18use std::pin::Pin;
19use std::sync::Arc;
20
21use futures::Future;
22#[cfg(target_family = "unix")]
23use libloading::os::unix::Library;
24#[cfg(target_family = "windows")]
25use libloading::Library;
26
27/// A `NodeConstructor` creates a single [`Node`](`Node`).
28///
29/// The `record` holds the metadata associated with the Node while the `constructor` is the function
30/// defined by the user to create it (through the implementation of [`Source`](`Source`),
31/// [`Operator`](`Operator`), or [`Sink`](`Sink`)).
32///
33/// The `_library` is a reference over the dynamically loaded shared library. It can be `None` when
34/// the factory is created programmatically.
35pub(crate) struct NodeConstructor<Record, C: ConstructorFn> {
36    pub(crate) record: Record,
37    pub(crate) constructor: C,
38    _library: Option<Arc<Library>>,
39}
40/// `ConstructorFn` is a private trait that prevents us from associating any function to the
41/// `Constructor` of [`NodeConstructor`](`NodeConstructor`) struct.
42pub(crate) trait ConstructorFn {}
43
44/// `SourceFn` is the only signature we accept to construct a [`Source`](`crate::prelude::Source`).
45pub type SourceFn = fn(
46    Context,
47    Option<Configuration>,
48    Outputs,
49) -> Pin<Box<dyn Future<Output = Result<Arc<dyn Node>>> + Send>>;
50
51impl ConstructorFn for SourceFn {}
52
53/// `OperatorFn` is the only signature we accept to construct an [`Operator`](`crate::prelude::Operator`).
54pub type OperatorFn = fn(
55    Context,
56    Option<Configuration>,
57    Inputs,
58    Outputs,
59) -> Pin<Box<dyn Future<Output = Result<Arc<dyn Node>>> + Send>>;
60
61impl ConstructorFn for OperatorFn {}
62
63/// `SinkFn` is the only signature we accept to construct a [`Sink`](`crate::prelude::Sink`).
64pub type SinkFn = fn(
65    Context,
66    Option<Configuration>,
67    Inputs,
68) -> Pin<Box<dyn Future<Output = Result<Arc<dyn Node>>> + Send>>;
69
70impl ConstructorFn for SinkFn {}
71
72/// A `SourceConstructor` generates a [`Source`](`crate::prelude::Source`).
73pub(crate) type SourceConstructor = NodeConstructor<SourceRecord, SourceFn>;
74
75/// An `OperatorConstructor` generates a [`Operator`](`crate::prelude::Operator`).
76pub(crate) type OperatorConstructor = NodeConstructor<OperatorRecord, OperatorFn>;
77
78/// A `SinkConstructor` generates a [`Sink`](`crate::prelude::Sink`).
79pub(crate) type SinkConstructor = NodeConstructor<SinkRecord, SinkFn>;
80
81/// Dereferencing to the record allows for an easy access to the metadata of the node.
82impl<Record, C: ConstructorFn> Deref for NodeConstructor<Record, C> {
83    type Target = Record;
84
85    fn deref(&self) -> &Self::Target {
86        &self.record
87    }
88}
89
90impl<Record, C: ConstructorFn> NodeConstructor<Record, C> {
91    /// Creates a NodeFactory without a `library`.
92    ///
93    /// This function is intended for internal use in order to create a data flow programmatically.
94    pub(crate) fn new_static(record: Record, constructor: C) -> Self {
95        Self {
96            record,
97            constructor,
98            _library: None,
99        }
100    }
101
102    pub(crate) fn new_dynamic(record: Record, constructor: C, library: Arc<Library>) -> Self {
103        Self {
104            record,
105            constructor,
106            _library: Some(library),
107        }
108    }
109}
zenoh_flow/runtime/dataflow/node.rs

zenoh_flow/runtime/dataflow/
node.rs
