Struct Runtime

Source

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

Expand description

Create a new runtime instance.

Fields§

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

Implementations§

Source §

impl Runtime

Source

pub async fn new( plugins: impl AsyncFnOnce(&mut FileExtManagerBuilder, &mut UrlSchemeManagerBuilder) -> Result<()>, ) -> Result<Self>

Create a new runtime instance with plugins.

Examples found in repository ?

examples/simple_runtime.rs (lines 43-47)

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

More examples

Hide additional examples

examples/service_runtime.rs (lines 39-43)

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

Source

pub async fn run( self, flows: Flows, nodes: impl AsyncFnOnce(&mut NodeLoader) -> Result<()>, ) -> Result<()>

Load all nodes with the flows provided and run them all.

Examples found in repository ?

examples/simple_runtime.rs (lines 51-65)

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

More examples

Hide additional examples

examples/service_runtime.rs (lines 50-63)

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