flarrow-api 0.3.1

flarrow (flow + arrow) is a rust runtime/framework for building dataflow applications.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

use flarrow_api::prelude::{thirdparty::*, *};

#[derive(Node)]
pub struct MyService {
    pub compare_to_128: Queryable<u8, String>,
    pub compare_to_64: Queryable<u8, String>,
}

#[node(runtime = "default_runtime")]
impl Node for MyService {
    async fn new(
        _: Inputs,
        _: Outputs,
        _: Queries,
        mut queryables: Queryables,
        _: serde_yml::Value,
    ) -> Result<Self> {
        Ok(Self {
            compare_to_128: queryables
                .with("compare_to_128")
                .await
                .wrap_err("Failed to create compare_to_128 queryable")?,
            compare_to_64: queryables
                .with("compare_to_64")
                .await
                .wrap_err("Failed to create compare_to_64 queryable")?,
        })
    }

    async fn start(self: Box<Self>) -> Result<()> {
        let mut compare_to_128 = self.compare_to_128;
        let task_128: tokio::task::JoinHandle<Result<()>> = tokio::spawn(async move {
            while let Ok(()) = compare_to_128
                .on_demand(async |query| match query > 128 {
                    true => Ok(format!("{} is greater than 128", query).to_string()),
                    false => Ok(format!("{} is less than or equal to 128", query).to_string()),
                })
                .await
            {}

            Ok(())
        });

        let mut compare_to_64 = self.compare_to_64;
        let task_64: tokio::task::JoinHandle<Result<()>> = tokio::spawn(async move {
            while let Ok(()) = compare_to_64
                .on_demand(async |query| match query > 64 {
                    true => Ok(format!("{} is greater than 64", query).to_string()),
                    false => Ok(format!("{} is less than or equal to 64", query).to_string()),
                })
                .await
            {}

            Ok(())
        });

        task_128.await??;
        task_64.await??;

        Ok(())
    }
}