Struct DataflowRuntime

Source

pub struct DataflowRuntime {
    pub clock: Arc<HLC>,
    pub nodes: HashMap<NodeID, RuntimeNode>,
}

Fields§

§clock: Arc<HLC>§nodes: HashMap<NodeID, RuntimeNode>

Implementations§

Source §

impl DataflowRuntime

Source

pub async fn new( flows: Flows, url_plugin: Option<RuntimeUrlPlugin>, load: impl AsyncFn(&mut Loader) -> Result<()>, ) -> Result<Self>

Examples found in repository ?

examples/runtime.rs (lines 73-92)

46async fn main() -> Result<()> {
47    let mut layout = DataflowLayout::new();
48
49    let (source, output) = layout
50        .create_node(async |io: &mut NodeIO| io.open_output("out"))
51        .await;
52
53    let (operator, (op_in, op_out)) = layout
54        .create_node(async |io: &mut NodeIO| (io.open_input("in"), io.open_output("out")))
55        .await;
56
57    let (sink, input) = layout
58        .create_node(async |io: &mut NodeIO| io.open_input("in"))
59        .await;
60
61    let layout = Arc::new(layout);
62    let flows = Flows::new(layout.clone(), async move |connector: &mut Connector| {
63        connector.connect(op_in, output)?;
64        connector.connect(input, op_out)?;
65
66        Ok(())
67    })
68    .await?;
69
70    let path = std::env::var("CARGO_MANIFEST_DIR")?;
71    let examples = format!("file://{}/../../target/debug/examples", path);
72
73    let runtime = DataflowRuntime::new(flows, None, async move |loader: &mut Loader| {
74        loader
75            .load_statically_linked::<MyOperator>(operator, serde_yml::Value::from(""))
76            .await
77            .wrap_err("Failed to load MyOperator")?;
78
79        let source_file = Url::parse("builtin:///timer")?;
80        let sink_file = Url::parse(&format!("{}/libsink.so", examples))?;
81
82        loader
83            .load_from_url(source, source_file, serde_yml::from_str("frequency: 5.0")?)
84            .await
85            .wrap_err("Failed to load source")?;
86        loader
87            .load_from_url(sink, sink_file, serde_yml::Value::from(""))
88            .await
89            .wrap_err("Failed to load sink")?;
90
91        Ok(())
92    })
93    .await?;
94
95    runtime.run().await
96}

Source

pub async fn run(self) -> Result<()>

Examples found in repository ?

examples/runtime.rs (line 95)

46async fn main() -> Result<()> {
47    let mut layout = DataflowLayout::new();
48
49    let (source, output) = layout
50        .create_node(async |io: &mut NodeIO| io.open_output("out"))
51        .await;
52
53    let (operator, (op_in, op_out)) = layout
54        .create_node(async |io: &mut NodeIO| (io.open_input("in"), io.open_output("out")))
55        .await;
56
57    let (sink, input) = layout
58        .create_node(async |io: &mut NodeIO| io.open_input("in"))
59        .await;
60
61    let layout = Arc::new(layout);
62    let flows = Flows::new(layout.clone(), async move |connector: &mut Connector| {
63        connector.connect(op_in, output)?;
64        connector.connect(input, op_out)?;
65
66        Ok(())
67    })
68    .await?;
69
70    let path = std::env::var("CARGO_MANIFEST_DIR")?;
71    let examples = format!("file://{}/../../target/debug/examples", path);
72
73    let runtime = DataflowRuntime::new(flows, None, async move |loader: &mut Loader| {
74        loader
75            .load_statically_linked::<MyOperator>(operator, serde_yml::Value::from(""))
76            .await
77            .wrap_err("Failed to load MyOperator")?;
78
79        let source_file = Url::parse("builtin:///timer")?;
80        let sink_file = Url::parse(&format!("{}/libsink.so", examples))?;
81
82        loader
83            .load_from_url(source, source_file, serde_yml::from_str("frequency: 5.0")?)
84            .await
85            .wrap_err("Failed to load source")?;
86        loader
87            .load_from_url(sink, sink_file, serde_yml::Value::from(""))
88            .await
89            .wrap_err("Failed to load sink")?;
90
91        Ok(())
92    })
93    .await?;
94
95    runtime.run().await
96}