core_lib/exec/
runner.rs

1#![allow(dead_code)]
2use std::{collections::HashMap, sync::Arc};
3
4use anyhow::Result;
5use tokio::sync::Mutex;
6
7use crate::{
8    config::{Cmd, Job, ProjectConfig},
9    core::watcher::WatchContext,
10    exec::{
11        OutpuStrategy, PipeRegistry,
12        command::{CommandOutput, exec_background, exec_timeout},
13        container::contain_cmd,
14    },
15};
16
17const DEFAULT_TIMEOUT: u64 = 300;
18pub struct JobNode {
19    pub job: Arc<Job>,
20    pub depend_on: Vec<String>, // names of jobs this job depends on
21    pub remaining_dependencies: usize,
22    pub dependents: Vec<String>, // names of jobs that depend on this one
23}
24
25pub fn build_dependency_graph(config: &ProjectConfig) -> Result<HashMap<String, JobNode>> {
26    let mut graph = HashMap::new();
27
28    for (name, job) in config.pipeline.jobs.iter() {
29        if job.needs.contains(name) {
30            return Err(anyhow::anyhow!("Job: {} cannot depend on itself", name));
31        }
32
33        let job_needs = if !job.pipe.is_empty() && !job.needs.contains(&job.pipe) {
34            let mut j = job.needs.clone();
35            j.push(job.pipe.clone());
36            j
37        } else {
38            job.needs.clone()
39        };
40
41        let node = JobNode {
42            job: Arc::new(job.clone()),
43            depend_on: job_needs.clone(),
44            remaining_dependencies: job_needs.len(),
45            dependents: vec![],
46        };
47        graph.insert(name.clone(), node);
48    }
49
50    for name in graph.keys().cloned().collect::<Vec<_>>() {
51        let depend_on = graph[&name].depend_on.clone();
52        for dep in depend_on {
53            if let Some(dep_node) = graph.get_mut(&dep) {
54                dep_node.dependents.push(name.clone());
55            } else {
56                anyhow::bail!("Job '{}' depends on unknown job '{}'", name, dep);
57            }
58        }
59    }
60    Ok(graph)
61}
62
63pub async fn run_step(
64    ctx: &WatchContext,
65    step: &Cmd,
66    env: &Option<HashMap<String, String>>,
67    output_strategy: &OutpuStrategy,
68    pipe_registry: Arc<Mutex<PipeRegistry>>,
69) -> Result<Option<CommandOutput>> {
70    let parts = shell_words::split(&step.cmd)?;
71
72    if let Some(container) = &step.container {
73        contain_cmd(
74            container,
75            parts,
76            env.clone(),
77            &ctx.project_dir,
78            &ctx.log_path().display().to_string(),
79            Some(ctx.config.timeout.unwrap_or(DEFAULT_TIMEOUT)),
80        )
81        .await?;
82    } else if step.blocking {
83        background_process(ctx, parts, env.clone()).await?;
84    } else {
85        return Ok(Some(
86            timeout_process(
87                ctx,
88                parts,
89                env.clone(),
90                ctx.config.timeout.unwrap_or(DEFAULT_TIMEOUT),
91                output_strategy,
92                pipe_registry,
93            )
94            .await?,
95        ));
96    }
97    Ok(None)
98}
99
100async fn background_process(
101    ctx: &WatchContext,
102    parts: Vec<String>,
103    env: Option<HashMap<String, String>>,
104) -> Result<(), anyhow::Error> {
105    match exec_background(parts.clone(), ctx, env).await {
106        Ok(_) => {}
107        Err(e) => {
108            return Err(e);
109        }
110    };
111    Ok(())
112}
113
114async fn timeout_process(
115    ctx: &WatchContext,
116    parts: Vec<String>,
117    env: Option<HashMap<String, String>>,
118    default_timeout: u64,
119    output_strategy: &OutpuStrategy,
120    pipe_registry: Arc<Mutex<PipeRegistry>>,
121) -> Result<CommandOutput, anyhow::Error> {
122    match exec_timeout(
123        parts.clone(),
124        ctx,
125        default_timeout,
126        env,
127        output_strategy,
128        pipe_registry,
129    )
130    .await
131    {
132        Ok(o) => Ok(o),
133        Err(e) => Err(e),
134    }
135}
core_lib/exec/runner.rs

core_lib/exec/
runner.rs
