zenoh-flow 0.5.0-alpha.2

Zenoh-Flow: a Zenoh-based data flow programming framework for computations that span from the cloud to the device.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

//
// Copyright (c) 2021 - 2023 ZettaScale Technology
//
// This program and the accompanying materials are made available under the
// terms of the Eclipse Public License 2.0 which is available at
// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
// which is available at https://www.apache.org/licenses/LICENSE-2.0.
//
// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
//
// Contributors:
//   ZettaScale Zenoh Team, <zenoh@zettascale.tech>
//

use crate::model::record::{OperatorRecord, SinkRecord, SourceRecord};
use crate::prelude::{Configuration, Context, Inputs, Node, Outputs, Result};
use std::ops::Deref;
use std::pin::Pin;
use std::sync::Arc;

use futures::Future;
#[cfg(target_family = "unix")]
use libloading::os::unix::Library;
#[cfg(target_family = "windows")]
use libloading::Library;

/// A `NodeConstructor` creates a single [`Node`](`Node`).
///
/// The `record` holds the metadata associated with the Node while the `constructor` is the function
/// defined by the user to create it (through the implementation of [`Source`](`Source`),
/// [`Operator`](`Operator`), or [`Sink`](`Sink`)).
///
/// The `_library` is a reference over the dynamically loaded shared library. It can be `None` when
/// the factory is created programmatically.
pub(crate) struct NodeConstructor<Record, C: ConstructorFn> {
    pub(crate) record: Record,
    pub(crate) constructor: C,
    _library: Option<Arc<Library>>,
}
/// `ConstructorFn` is a private trait that prevents us from associating any function to the
/// `Constructor` of [`NodeConstructor`](`NodeConstructor`) struct.
pub(crate) trait ConstructorFn {}

/// `SourceFn` is the only signature we accept to construct a [`Source`](`crate::prelude::Source`).
pub type SourceFn = fn(
    Context,
    Option<Configuration>,
    Outputs,
) -> Pin<Box<dyn Future<Output = Result<Arc<dyn Node>>> + Send>>;

impl ConstructorFn for SourceFn {}

/// `OperatorFn` is the only signature we accept to construct an [`Operator`](`crate::prelude::Operator`).
pub type OperatorFn = fn(
    Context,
    Option<Configuration>,
    Inputs,
    Outputs,
) -> Pin<Box<dyn Future<Output = Result<Arc<dyn Node>>> + Send>>;

impl ConstructorFn for OperatorFn {}

/// `SinkFn` is the only signature we accept to construct a [`Sink`](`crate::prelude::Sink`).
pub type SinkFn = fn(
    Context,
    Option<Configuration>,
    Inputs,
) -> Pin<Box<dyn Future<Output = Result<Arc<dyn Node>>> + Send>>;

impl ConstructorFn for SinkFn {}

/// A `SourceConstructor` generates a [`Source`](`crate::prelude::Source`).
pub(crate) type SourceConstructor = NodeConstructor<SourceRecord, SourceFn>;

/// An `OperatorConstructor` generates a [`Operator`](`crate::prelude::Operator`).
pub(crate) type OperatorConstructor = NodeConstructor<OperatorRecord, OperatorFn>;

/// A `SinkConstructor` generates a [`Sink`](`crate::prelude::Sink`).
pub(crate) type SinkConstructor = NodeConstructor<SinkRecord, SinkFn>;

/// Dereferencing to the record allows for an easy access to the metadata of the node.
impl<Record, C: ConstructorFn> Deref for NodeConstructor<Record, C> {
    type Target = Record;

    fn deref(&self) -> &Self::Target {
        &self.record
    }
}

impl<Record, C: ConstructorFn> NodeConstructor<Record, C> {
    /// Creates a NodeFactory without a `library`.
    ///
    /// This function is intended for internal use in order to create a data flow programmatically.
    pub(crate) fn new_static(record: Record, constructor: C) -> Self {
        Self {
            record,
            constructor,
            _library: None,
        }
    }

    pub(crate) fn new_dynamic(record: Record, constructor: C, library: Arc<Library>) -> Self {
        Self {
            record,
            constructor,
            _library: Some(library),
        }
    }
}