flarrow_runtime/
runtime.rs

1use std::{collections::HashMap, sync::Arc};
2
3use crate::prelude::*;
4
5/// Create a new runtime instance.
6pub struct Runtime {
7    pub clock: Arc<HLC>,
8
9    pub file_ext: Arc<FileExtManager>,
10    pub url_scheme: Arc<UrlSchemeManager>,
11
12    pub nodes: HashMap<NodeLayout, RuntimeNode>,
13}
14
15impl Runtime {
16    /// Create a new runtime instance with plugins.
17    pub async fn new(
18        plugins: impl AsyncFnOnce(
19            &mut FileExtManagerBuilder,
20            &mut UrlSchemeManagerBuilder,
21        ) -> Result<()>,
22    ) -> Result<Self> {
23        let mut file_ext = FileExtManagerBuilder::new().await?;
24        let mut url_scheme = UrlSchemeManagerBuilder::new().await?;
25
26        plugins(&mut file_ext, &mut url_scheme).await?;
27
28        Ok(Self {
29            clock: Arc::new(HLC::default()),
30            file_ext: Arc::new(FileExtManager::new(file_ext.plugins)),
31            url_scheme: Arc::new(UrlSchemeManager::new(url_scheme.plugins)),
32            nodes: HashMap::new(),
33        })
34    }
35
36    /// Load all nodes with the flows provided and run them all.
37    pub async fn run(
38        mut self,
39        flows: Flows,
40        nodes: impl AsyncFnOnce(&mut NodeLoader) -> Result<()>,
41    ) -> Result<()> {
42        let mut node_loader =
43            NodeLoader::new(self.file_ext, self.url_scheme, self.clock.clone(), flows);
44
45        nodes(&mut node_loader).await?;
46
47        self.nodes.extend(node_loader.nodes);
48
49        println!("Starting runtime... (press Ctrl+C to stop)");
50
51        let mut tasks = Vec::new();
52        for (layout, node) in self.nodes {
53            tasks.push(tokio::spawn(async move {
54                node.run().await.wrap_err(format!(
55                    "Node '{}' (uuid: {}) failed",
56                    layout.label, layout.uuid,
57                ))
58            }));
59        }
60
61        let join_all = tokio::spawn(async move {
62            let mut reports: Vec<eyre::Report> = Vec::new();
63
64            let mut is_ok = true;
65
66            for task in tasks {
67                let result = task.await?;
68
69                if let Err(report) = result {
70                    is_ok = false;
71                    reports.push(report);
72                }
73            }
74
75            let reports: eyre::Report = {
76                let report_str: String = reports.iter().fold(
77                    "The runtime encountered multiple errors:".to_string(),
78                    |acc, report| format!("{}\n\n{:?}", acc, report),
79                );
80
81                eyre::Report::msg(report_str)
82            };
83
84            match is_ok {
85                true => Ok(()),
86                false => Err(reports),
87            }
88        });
89
90        tokio::select! {
91            _ = tokio::signal::ctrl_c() => { Ok(()) },
92            results = join_all => { results? }
93        }
94    }
95}
flarrow_runtime/runtime.rs

flarrow_runtime/
runtime.rs
