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

use std::{
    io::{BufRead, BufReader},
    process::{Command, Stdio},
    thread,
    time::Duration,
};

use chrono::Utc;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use task_options::TaskOptions;
use task_result::TaskResult;
use thepipelinetool_utils::{value_from_file, value_to_file};

pub mod branch;
pub mod ordered_queued_task;
pub mod queued_task;
pub mod task_options;
pub mod task_ref_inner;
pub mod task_result;
pub mod task_status;

pub const DAGS_DIR: &str = "./bin";

#[derive(Clone, Serialize, Deserialize, Debug)]
pub struct Task {
    pub id: usize,
    pub function_name: String,
    pub template_args: Value,
    pub options: TaskOptions,
    pub lazy_expand: bool,
    pub is_dynamic: bool,
    pub is_branch: bool,
}

impl Task {
    pub fn execute(
        &self,
        resolved_args: &Value,
        attempt: usize,
        handle_stdout: Box<dyn Fn(String) + Send>,
        handle_stderr: Box<dyn Fn(String) + Send>,
        executable_path: &str,
    ) -> TaskResult {
        if attempt > 1 {
            thread::sleep(self.options.retry_delay);
        }

        let task_id: usize = self.id;
        let function_name = &self.function_name;
        let resolved_args_str = serde_json::to_string(resolved_args).unwrap();
        let in_path = format!("./in_{function_name}_{task_id}.json");
        let out_path = format!("./{function_name}_{task_id}.json");
        let use_timeout = self.options.timeout.is_some();
        let timeout_as_secs = self
            .options
            .timeout
            .unwrap_or(Duration::ZERO)
            .as_secs()
            .to_string();

        value_to_file(resolved_args, &in_path);

        let start = Utc::now();
        let mut child = Command::new(if use_timeout {
            "timeout"
        } else {
            executable_path
        })
        .args(if use_timeout {
            vec![
                "-k",
                &timeout_as_secs,
                &timeout_as_secs,
                executable_path,
                "run",
                "function",
                &function_name,
                &out_path,
                &in_path,
            ]
        } else {
            vec![
                "run",
                "function",
                function_name.as_str(),
                &out_path,
                &in_path,
            ]
        })
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .expect("failed to start command");

        let stdout = child.stdout.take().expect("failed to take stdout");
        let stderr = child.stderr.take().expect("failed to take stderr");

        // Spawn a thread to handle stdout
        let stdout_handle = thread::spawn(move || {
            let reader = BufReader::new(stdout);
            for line in reader.lines() {
                let line = format!("{}\n", line.expect("failed to read line from stdout"));
                handle_stdout(line);
            }
        });

        // Spawn a thread to handle stderr
        let stderr_handle = thread::spawn(move || {
            let reader = BufReader::new(stderr);
            for line in reader.lines() {
                let line = format!("{}\n", line.expect("failed to read line from stdout"));
                handle_stderr(line);
            }
        });

        let status = child.wait().expect("failed to wait on child");
        let end = Utc::now();

        let timed_out = matches!(status.code(), Some(124));

        // Join the stdout and stderr threads
        stdout_handle.join().expect("stdout thread panicked");
        stderr_handle.join().expect("stderr thread panicked");

        TaskResult {
            task_id,
            result: if status.success() {
                value_from_file(&out_path)
            } else {
                Value::Null
            },
            attempt,
            max_attempts: self.options.max_attempts,
            function_name: function_name.to_string(),
            resolved_args_str,
            success: status.success(),
            started: start.to_rfc3339(),
            ended: end.to_rfc3339(),
            elapsed: end.timestamp() - start.timestamp(),
            premature_failure: false,
            premature_failure_error_str: if timed_out {
                "timed out".into()
            } else {
                "".into()
            },
            is_branch: self.is_branch,
        }
    }
}