rskit-process 0.2.0-alpha.5

Process and subprocess execution with timeout and signal handling
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

use std::io::ErrorKind;
use std::process::Stdio;

use tokio::{process::ChildStdin, task::JoinHandle};

use crate::{
    AppError, AppResult, CapturedIo, ErrorCode, InputPolicy, ObservedIo, OutputPolicy,
    ProcessConfig,
};

pub(super) trait PipeIo {
    fn input(&self) -> &InputPolicy;
    fn output(&self) -> &OutputPolicy;
}

impl PipeIo for CapturedIo {
    fn input(&self) -> &InputPolicy {
        &self.input
    }

    fn output(&self) -> &OutputPolicy {
        &self.output
    }
}

impl PipeIo for ObservedIo {
    fn input(&self) -> &InputPolicy {
        &self.input
    }

    fn output(&self) -> &OutputPolicy {
        &self.output
    }
}

pub(super) fn stdin_stdio(input: &InputPolicy) -> Stdio {
    match input {
        InputPolicy::Closed => Stdio::null(),
        InputPolicy::Bytes(_) => Stdio::piped(),
        InputPolicy::Inherit => Stdio::inherit(),
    }
}

pub(super) fn pipe_stdin_stdio(input: &InputPolicy) -> AppResult<Stdio> {
    match input {
        InputPolicy::Closed => Ok(Stdio::null()),
        InputPolicy::Bytes(_) => Ok(Stdio::piped()),
        InputPolicy::Inherit => Err(AppError::invalid_input(
            "process.io.input",
            "inherited stdin requires inherited I/O mode; pipe-backed interactive stdin is not supported",
        )),
    }
}

pub(super) fn inherited_config(config: &ProcessConfig) -> ProcessConfig {
    let mut config = config.clone();
    config.signal = config
        .signal
        .with_create_process_group(false)
        .with_terminate_descendants(false);
    config
}

pub(super) fn spawn_stdin_writer(
    stdin: Option<ChildStdin>,
    input: &InputPolicy,
) -> Option<JoinHandle<AppResult<()>>> {
    let InputPolicy::Bytes(bytes) = input else {
        return None;
    };
    let mut stdin = stdin?;
    let bytes = bytes.clone();
    Some(tokio::spawn(async move {
        use tokio::io::AsyncWriteExt;

        match stdin.write_all(&bytes).await {
            Ok(()) => Ok(()),
            Err(error) if error.kind() == ErrorKind::BrokenPipe => Ok(()),
            Err(error) => Err(AppError::new(
                ErrorCode::Internal,
                format!("failed to write to stdin: {error}"),
            )),
        }
    }))
}