dag-flow 0.1.6

DAG Flow is a simple DAG workflow engine.
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

use std::collections::HashMap;
use std::iter;
use std::time::Duration;
use std::time::Instant;

use dag_flow::context::Context;
use dag_flow::engine::Engine;
use dag_flow::task::Input;
use dag_flow::task::Task;
use futures::StreamExt;
use futures::executor;
use futures::future::OptionFuture;
use futures::stream::FuturesUnordered;
use futures_timer::Delay;

const NUMBERS: &[u64] = &[1, 2, 3];

fn main() {
    let builder = Engine::builder();
    for &number in NUMBERS {
        builder.add_task(Square::from(number));
    }
    builder.add_task(Sum::from(NUMBERS.into()));

    let engine = builder.build().unwrap();
    let context = Context::new();

    let now = Instant::now();
    executor::block_on(engine.run(context.clone()));

    if let Some(&max) = NUMBERS.iter().max() {
        assert_eq!(now.elapsed().as_secs(), max);
    }

    let ids: Vec<_> = NUMBERS
        .iter()
        .copied()
        .map(Square::id)
        .chain(iter::once(Sum::id()))
        .collect();

    let data: HashMap<_, _> = executor::block_on(
        ids.iter()
            .map(|id| {
                let data = context.get(id);
                async move { (id, OptionFuture::from(data).await.unwrap_or_default()) }
            })
            .collect::<FuturesUnordered<_>>()
            .collect(),
    );

    for (id, &number) in ids.iter().zip(NUMBERS) {
        assert_eq!(data[id], Some(number.pow(2)));
    }
}

struct Square {
    id: String,
    number: u64,
}

impl Square {
    fn id(number: u64) -> String {
        format!("square-{number}")
    }

    fn from(number: u64) -> Self {
        Self {
            id: Self::id(number),
            number,
        }
    }
}

impl Task<String, u64> for Square {
    fn id(&self) -> String {
        self.id.clone()
    }

    async fn run(&self, _: HashMap<String, Input<'_, u64>>) -> Option<u64> {
        Delay::new(Duration::from_secs(self.number)).await;
        Some(self.number.pow(2))
    }
}

struct Sum {
    id: String,
    dependencies: Vec<String>,
    numbers: Vec<u64>,
}

impl Sum {
    fn id() -> String {
        "sum".into()
    }

    fn from(numbers: Vec<u64>) -> Self {
        Self {
            id: Self::id(),
            dependencies: numbers.iter().copied().map(Square::id).collect(),
            numbers,
        }
    }
}

impl Task<String, u64> for Sum {
    fn id(&self) -> String {
        self.id.clone()
    }

    fn dependencies(&self) -> Vec<String> {
        self.dependencies.clone()
    }

    async fn run(&self, inputs: HashMap<String, Input<'_, u64>>) -> Option<u64> {
        self.numbers
            .iter()
            .enumerate()
            .map(|(index, _)| inputs[&self.dependencies[index]].clone())
            .collect::<FuturesUnordered<_>>()
            .collect::<Vec<_>>()
            .await
            .into_iter()
            .sum()
    }
}