willdo 0.0.1

Task manager with DAG
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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153

//! Not totally simple job runner

use crate::execution::BoxError;
use crate::execution::{
    progress::Observation, repository::Execution, BoxFuture, BoxStream, JobRepository, Progress,
    Runner,
};
use crate::job::JobId;
use core::{pin::Pin, task::{Context, Poll}};
use futures_lite::{FutureExt, Stream, StreamExt as _};
use std::collections::VecDeque;

/// A not-completely-simple runner will run all required jobs to completion.
///
/// It will run alternative paths in parallel and will skip unnecessary jobs.
///
/// It only reports basic start and completion of jobs (that's why it's simple).
pub struct Simple;

impl Runner for Simple {
    fn run<'f, 'j: 'f, J: 'j + JobRepository + Send + Unpin>(
        &self,
        jobs: J,
        goals: &[Box<str>],
    ) -> BoxStream<'f, (JobId, Progress)> {
        let goals = goals.to_vec();
        // let ready = jobs
        //     .select(&goals)
        //     .into_iter()
        //     .map(|j| (j, Progress::Select))
        //     .collect();
        let ready = vec![].into();
        Box::pin(SimpleEvents {
            jobs,
            goals,
            running: vec![],
            ready,
            reload: false,
        })
    }
}

struct SimpleEvents<J> {
    jobs: J,
    goals: Vec<Box<str>>,
    running: Vec<BoxFuture<'static, Execution>>,
    ready: VecDeque<(JobId, Progress)>,
    reload: bool,
}
impl<J> Stream for SimpleEvents<J>
where
    J: JobRepository + Send + Unpin,
{
    type Item = (JobId, Progress);
    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let Self {
            jobs,
            goals,
            running,
            ready,
            reload,
        } = self.as_mut().get_mut();

        loop {
            if let Some(ready) = ready.pop_front() {
                break Poll::Ready(Some(ready));
            }

            if *reload || running.is_empty() {
                *reload = false;
                let runnable = jobs.execute(goals);
                ready.extend(
                    runnable
                        .iter()
                        .map(|e| (e.job().id.clone(), Progress::Start)),
                );
                running.extend(runnable.into_iter().map(|mut e| {
                    Box::pin(async move {
                        run(&mut e).await;
                        e
                    }) as BoxFuture<'static, Execution>
                }));
                if running.is_empty() {
                    break Poll::Ready(None);
                } else {
                    continue;
                }
            }
            let mut completed = None;
            for (i, v) in running.iter_mut().enumerate() {
                match v.poll(cx) {
                    Poll::Ready(execution) => {
                        completed = Some((i, execution));
                        break;
                    }
                    Poll::Pending => continue,
                }
            }
            let Some((i, execution)) = completed else {
                return Poll::Pending;
            };
            *reload = true;
            drop(running.remove(i));
            ready.push_back((
                execution.job().id.clone(),
                Progress::Observation(Observation::Completed(
                    execution.progress().get_status().unwrap_or(1),
                )),
            ));
        }
    }
}

/// Run a single execution.
async fn run(execution: &mut Execution) {
    if let Err(e) = run_job_fallible(execution).await {
        execution
            .progress()
            .update(Progress::Observation(Observation::Failure {
                message: "job execution failed".into(),
                source: Some(e.into()),
            }));
    }
}

async fn run_job_fallible(
    execution: &mut Execution,
) -> Result<(), BoxError> {
    let script = &execution.job().script.clone();
    let mut interpretter = execution.commander().start()?;
    execution.progress().update(Progress::Start);
    for command in script {
        execution
            .progress()
            .update(Progress::Command(command.clone()));
        let mut logs = interpretter.execute(command).await?;
        while let Some(observation) = logs.next().await {
            execution
                .progress()
                .update(Progress::Observation(observation));
        }
        if execution.progress().is_failed() {
            break;
        }
    }
    execution.progress().update(Progress::Shutdown);
    let mut shutdown = interpretter.shutdown().await?;
    while let Some(log) = shutdown.next().await {
        execution.progress().update(Progress::Observation(log));
    }

    Ok(())
}