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
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

pub mod message;
pub mod rolling_buffer;

use crate::runtime::{ctx::Ctx, job::task::action::OptionalAction};
use anyhow::{Context, Result};
use crossbeam_channel::{unbounded, Sender};
use log::debug;
use message::{LogMonitorCmd, LogMonitorMessage};
use rolling_buffer::RollingBuffer;
use std::process::{Command, Stdio};
use std::thread;

/// Represents and contains a given runtime job task log monitor.
///
/// This object contains all of the data necessary to run a given log monitor. This data includes
/// the log monitor name, the size of its rolling buffer, the rolling buffer instance itself, the
/// `test` command which should be executed on each push to the buffer, and the `ontrigger` action
/// which should run if `test` returns with a `0` exit code.
#[derive(Clone, Debug)]
pub struct LogMonitor {
    pub buffer: RollingBuffer,
    pub buffer_size: usize,
    pub ctx: Ctx,
    pub name: String,
    pub ontrigger: String,
    pub test: String,
}

impl LogMonitor {
    /// Constructs a new, empty `LogMonitor`.
    pub fn new(name: String) -> Self {
        Self {
            buffer: RollingBuffer::new(20),
            buffer_size: 20,
            ctx: Ctx::new(),
            name,
            ontrigger: String::new(),
            test: String::new(),
        }
    }

    /// Builds `LogMonitor` with the specified rolling buffer size.
    pub fn buffer_size(mut self, b: usize) -> Self {
        self.buffer = RollingBuffer::new(b);
        self.buffer_size = b;

        self
    }

    /// Builds `LogMonitor` with the name of the action to execute if the `test` succeeds.
    pub fn ontrigger(mut self, o: String) -> Self {
        self.ontrigger = o;

        self
    }

    /// Builds `LogMonitor` with the specified test.
    pub fn test(mut self, t: String) -> Self {
        self.test = t;

        self
    }

    /// Executes the log monitor using the provided action.
    pub fn run(
        mut self,
        ontrigger: OptionalAction,
    ) -> Result<(thread::JoinHandle<()>, Sender<LogMonitorMessage>)> {
        debug!("Running log_monitor instance with structure:\n{:#?}", self);

        let name = self.name.clone();

        let (sender, receiver) = unbounded::<LogMonitorMessage>();

        let handle = thread::Builder::new()
            .name(name.clone())
            .spawn(move || {
                debug!("Spawned log_monitor thread \"{}\"", &name);

                loop {
                    if let Ok(LogMonitorMessage { cmd, message }) = receiver.recv() {
                        debug!("Received message: {:?}", message);

                        match cmd {
                            LogMonitorCmd::Close => {
                                debug!("Received close message.");
                                break;
                            }
                            LogMonitorCmd::Log => {
                                self.push(message, &ontrigger);
                            }
                            LogMonitorCmd::None => debug!("Received empty message."),
                        }
                    }
                }

                debug!("Closing log_monitor thread \"{}\"", &name);
            })
            .context("Error spawning log monitor thread")?;

        Ok((handle, sender))
    }

    /// Pushes a line of text to the rolling buffer and executes the test command on the new buffer
    /// state.
    pub fn push(&mut self, line: String, ontrigger: &OptionalAction) {
        self.buffer.push(line);
        self.exec_test(ontrigger).ok();
    }

    /// Executes the current test command and, if successful, performs the `ontrigger` action.
    pub fn exec_test(&self, ontrigger: &OptionalAction) -> Result<()> {
        let bin = self.ctx.bin_command.bin.clone();
        let mut bin_args = self.ctx.bin_command.args.clone();
        bin_args.push(self.test.clone());

        let status = Command::new(bin)
            .args(bin_args)
            .env("ARPX_BUFFER", &self.buffer.dump()[..])
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .context(format!(
                "Error spawning test command on log monitor \"{}\"",
                self.name
            ))?
            .wait()
            .context(format!(
                "Error waiting for test command child on log monitor \"{}\"",
                self.name
            ))?;

        if status.success() {
            debug!("LogMonitor {} triggered", self.name);

            if let Some(action) = ontrigger {
                debug!(
                    "Running ontrigger \"{}\" from prepared action",
                    self.ontrigger
                );

                action();
            }
        }

        Ok(())
    }
}