arpx 0.5.0

Small-scale process orchestration
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

pub mod action;
pub mod log_monitor;
pub mod process;

use crate::runtime::ctx::Ctx;
use action::{get_log_monitor_action, get_process_actions};
use anyhow::{bail, Error, Result};
use log::debug;
use process::Process;
use std::thread;

/// Represents and contains a given runtime job task.
///
/// This object contains a list of processes which are executed in order when the task is run.
#[derive(Clone, Debug)]
pub struct Task {
    pub processes: Vec<Process>,
}

impl Task {
    /// Constructs a new, empty `Task`.
    pub fn new(processes: Vec<Process>) -> Self {
        Self { processes }
    }

    /// Executes defined processes in order.
    pub fn run(self, ctx: &Ctx) -> Result<()> {
        debug!("Running task instance with structure:\n{:#?}", self);

        let mut thread_handles = Vec::new();

        for process in self.processes {
            let mut log_monitor_senders = Vec::new();
            for log_monitor_name in &process.log_monitors {
                let log_monitor = &ctx.log_monitor_map[log_monitor_name];

                let log_monitor_action = get_log_monitor_action(log_monitor, ctx);
                let (handle, sender) = log_monitor.clone().run(log_monitor_action)?;

                thread_handles.push(handle);
                log_monitor_senders.push(sender);
            }

            let cloned_ctx = ctx.clone();
            let process_handle = thread::Builder::new()
                .name(process.name.clone())
                .spawn(move || {
                    debug!("Spawned thread \"{}\"", process.name);

                    let process_actions = get_process_actions(&process, &cloned_ctx);
                    process
                        .run(process_actions, &cloned_ctx, &log_monitor_senders)
                        .ok();

                    debug!("Closing thread \"{}\"", process.name);
                })
                .map_err(Error::new)?;

            thread_handles.push(process_handle);
        }

        for handle in thread_handles {
            if handle.join().is_err() {
                bail!("Error joining thread handle");
            }
        }

        Ok(())
    }
}