Struct DataflowLayout

Source

pub struct DataflowLayout {
    pub inputs: HashSet<InputID>,
    pub outputs: HashSet<OutputID>,
}

Fields§

§inputs: HashSet<InputID>§outputs: HashSet<OutputID>

Implementations§

Source §

impl DataflowLayout

Source

pub fn new() -> DataflowLayout

Examples found in repository ?

examples/runtime.rs (line 47)

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 create_node<T>( &mut self, node: impl AsyncFn(&mut NodeIO) -> T, ) -> (NodeID, T)

Examples found in repository ?

examples/runtime.rs (line 50)

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}