Struct NodeLoader

Source

pub struct NodeLoader {
    pub file_ext: Arc<FileExtManager>,
    pub url_scheme: Arc<UrlSchemeManager>,
    pub clock: Arc<HLC>,
    pub flows: Flows,
    pub nodes: HashMap<NodeLayout, RuntimeNode>,
}

Expand description

Loader struct passed to the user closure to load nodes

Fields§

§file_ext: Arc<FileExtManager>§url_scheme: Arc<UrlSchemeManager>§clock: Arc<HLC>§flows: Flows§nodes: HashMap<NodeLayout, RuntimeNode>

Implementations§

Source §

impl NodeLoader

Source

pub fn new( file_ext: Arc<FileExtManager>, url_scheme: Arc<UrlSchemeManager>, clock: Arc<HLC>, flows: Flows, ) -> Self

Source

pub async fn load<T: Node + 'static>( &mut self, source: NodeLayout, configuration: Value, ) -> Result<()>

Load a node from a Rust struct directly (statically linked)

Examples found in repository ?

examples/simple_runtime.rs (line 47)

4async fn main() -> Result<()> {
5    tracing_subscriber::fmt::init();
6
7    let mut layout = DataflowLayout::new();
8
9    let (source, output) = layout
10        .node("source", async |builder: &mut NodeIOBuilder| {
11            builder.output("out")
12        })
13        .await;
14
15    let (operator, (op_in, op_out)) = layout
16        .node("operator", async |builder: &mut NodeIOBuilder| {
17            (builder.input("in"), builder.output("out"))
18        })
19        .await;
20
21    let (sink, input) = layout
22        .node("sink", async |builder: &mut NodeIOBuilder| {
23            builder.input("in")
24        })
25        .await;
26
27    let layout = layout.build();
28
29    let flows = Flows::new(layout.clone(), async move |builder: &mut FlowsBuilder| {
30        builder.connect(op_in, output, None)?;
31        builder.connect(input, op_out, None)?;
32
33        Ok(())
34    })
35    .await?;
36
37    let runtime = Runtime::new(
38        async |_file_ext: &mut FileExtManagerBuilder, _url_scheme: &mut UrlSchemeManagerBuilder| {
39            Ok(())
40        },
41    )
42    .await?;
43
44    runtime
45        .run(flows, async move |loader: &mut NodeLoader| {
46            loader
47                .load::<Timer>(source, serde_yml::from_str("frequency: 1.0")?)
48                .await?;
49
50            loader
51                .load::<Transport>(operator, serde_yml::from_str("")?)
52                .await?;
53
54            loader
55                .load::<Printer>(sink, serde_yml::from_str("")?)
56                .await?;
57
58            Ok(())
59        })
60        .await
61}

Source

pub async fn load_url( &mut self, url: Url, source: NodeLayout, configuration: Value, ) -> Result<()>

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)

Examples found in repository ?

examples/service_runtime.rs (line 50)

4async fn main() -> Result<()> {
5    tracing_subscriber::fmt::init();
6
7    let mut layout = DataflowLayout::new();
8
9    let (service, (compare_to_128, compare_to_64)) = layout
10        .node("service", async |builder: &mut NodeIOBuilder| {
11            (
12                builder.queryable("compare_to_128"),
13                builder.queryable("compare_to_64"),
14            )
15        })
16        .await;
17
18    let (client, (ask_128, ask_64)) = layout
19        .node("client", async |builder: &mut NodeIOBuilder| {
20            (builder.query("ask_128"), builder.query("ask_64"))
21        })
22        .await;
23
24    let layout = layout.build();
25
26    let flows = Flows::new(layout.clone(), async move |builder: &mut FlowsBuilder| {
27        builder.connect(ask_128, compare_to_128, None)?;
28        builder.connect(ask_64, compare_to_64, None)?;
29
30        Ok(())
31    })
32    .await?;
33
34    let runtime = Runtime::new(
35        async |_file_ext: &mut FileExtManagerBuilder, _url_scheme: &mut UrlSchemeManagerBuilder| {
36            Ok(())
37        },
38    )
39    .await?;
40
41    let path = std::env::var("CARGO_MANIFEST_DIR")?;
42    let examples = format!("file://{}/../../target/debug/examples", path);
43
44    runtime
45        .run(flows, async move |loader: &mut NodeLoader| {
46            let service_file = Url::parse(&format!("{}/libservice.so", examples))?;
47            let client_file = Url::parse(&format!("{}/libclient.so", examples))?;
48
49            loader
50                .load_url(service_file, service, serde_yml::from_str("")?)
51                .await?;
52
53            loader
54                .load_url(client_file, client, serde_yml::from_str("")?)
55                .await?;
56
57            Ok(())
58        })
59        .await
60}