cb2_core 0.1.2

The core for the cb2 command runner
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

use crate::exec::FutureSig;
use crate::report::Report;
use crate::task::{RunMode, Task, TaskGroup};
use crate::task_item::task_item;
use crate::task_seq::task_seq;
use chrono::Utc;
use futures::future::lazy;
use futures::sync::mpsc::Sender;
use futures::Future;
use futures::Sink;

pub fn task_group(group: TaskGroup, sender: Sender<Report>) -> FutureSig {
    let id_clone = group.id.clone();
    let begin_clone = sender.clone();
    Box::new(lazy(move || {
        let items = group.items.into_iter().map(move |item| match item {
            Task::Item(item) => task_item(item, sender.clone()),
            Task::Group(group) => match group.run_mode {
                RunMode::Series => task_seq(group, sender.clone()),
                RunMode::Parallel => task_group(group, sender.clone()),
            },
        });

        let begin_time = Utc::now();

        begin_clone
            .clone()
            .send(Report::GroupStarted {
                id: id_clone.clone(),
                time: begin_time.clone(),
            })
            .then(move |_res: _| {
                futures::collect(items).then(move |res| {
                    let (items, all_valid) = match res.clone() {
                        Ok(items) => {
                            let valid = items.iter().all(|x| x.is_ok());
                            (items, valid)
                        }
                        Err(err_report) => (vec![Err(err_report)], false),
                    };

                    if all_valid {
                        Ok(Ok(Report::EndGroup {
                            time: Utc::now(),
                            id: id_clone,
                            dur: Utc::now().signed_duration_since(begin_time),
                            reports: items.clone(),
                        }))
                    } else {
                        Ok(Err(Report::ErrorGroup {
                            time: Utc::now(),
                            id: id_clone,
                            dur: Utc::now().signed_duration_since(begin_time),
                            reports: items.clone(),
                        }))
                    }
                })
            })
    }))
}