Struct Flows

Source

pub struct Flows {
    pub senders: Arc<Mutex<HashMap<OutputID, Sender<DataflowMessage>>>>,
    pub receivers: Arc<Mutex<HashMap<InputID, Receiver<DataflowMessage>>>>,
}

Fields§

§senders: Arc<Mutex<HashMap<OutputID, Sender<DataflowMessage>>>>§receivers: Arc<Mutex<HashMap<InputID, Receiver<DataflowMessage>>>>

Implementations§

Source §

impl Flows

Source

pub async fn new( layout: Arc<DataflowLayout>, flows: impl AsyncFn(&mut Connector) -> Result<()>, ) -> Result<Self>

Examples found in repository ?

examples/runtime.rs (lines 62-67)

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}