iridis 0.4.0

iridis is a framework that lets you define and build dataflow applications with ease.
Documentation 
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

//! This module defines the nodes `Loader`, used by the user
//! in order to load nodes in the runtime.

use std::{collections::HashMap, sync::Arc};

use crate::prelude::{thirdparty::tokio::task::JoinSet, *};

/// Loader struct passed to the user closure to load nodes
pub struct Loader {
    pub file_ext: Arc<FileExtManager>,
    pub url_scheme: Arc<UrlSchemeManager>,

    pub clock: Arc<HLC>,

    pub flows: RuntimeFlows,

    pub futures: JoinSet<Result<(NodeID, RuntimeNode)>>,
}

impl Loader {
    pub fn new(
        file_ext: Arc<FileExtManager>,
        url_scheme: Arc<UrlSchemeManager>,
        clock: Arc<HLC>,
        flows: RuntimeFlows,
    ) -> Self {
        Self {
            file_ext,
            url_scheme,
            clock,
            flows,
            futures: JoinSet::new(),
        }
    }

    /// Load a node from a Rust struct directly (statically linked)
    pub fn load<T: Node + 'static>(&mut self, source: NodeID, configuration: serde_yml::Value) {
        let (inputs, outputs, queries, queryables) = self
            .flows
            .node_primitives(self.clock.clone(), source.clone());

        self.futures.spawn(async move {
            let node = RuntimeNode::StaticallyLinked(
                T::new(inputs, outputs, queries, queryables, configuration)
                    .await?
                    .wrap_err(format!(
                        "Node '{}' (uuid: {}) failed to initialize",
                        source.label, source.uuid,
                    ))?,
            );

            tracing::debug!(
                "Node '{}' (uuid: {}) loaded from Rust struct {}",
                source.label,
                source.uuid,
                std::any::type_name::<T>()
            );

            Ok((source, node))
        });
    }

    /// Load a node from an URL. Be careful, you must ensure that the runtime has the necessary plugins to process this URL.
    /// By default you can pass all URL for the builtins nodes (builtin://) and all URL for dynamic libraries on the computer (file:///path/to/library.so)
    pub fn load_url(&mut self, url: Url, source: NodeID, configuration: serde_yml::Value) {
        let (inputs, outputs, queries, queryables) = self
            .flows
            .node_primitives(self.clock.clone(), source.clone());

        let file_ext = self.file_ext.clone();
        let url_scheme = self.url_scheme.clone();

        self.futures.spawn(async move {
            let node = url_scheme
                .load(
                    url.clone(),
                    inputs,
                    outputs,
                    queries,
                    queryables,
                    configuration,
                    file_ext,
                )
                .await?;

            tracing::debug!(
                "Node '{}' (uuid: {}) loaded from URL {:?}",
                source.label,
                source.uuid,
                url
            );

            Ok((source, node))
        });
    }

    /// `await` for all nodes to be loaded and return them
    pub async fn finish(mut self) -> Result<HashMap<NodeID, RuntimeNode>> {
        let mut nodes = HashMap::new();

        while let Some(res) = self.futures.join_next().await {
            let (source, node) = res??;

            nodes.insert(source, node);
        }

        Ok(nodes)
    }
}