rskit_process/persistent/
config.rs

1use std::time::Duration;
2
3use crate::{ProcessSpec, command::DEFAULT_MAX_OUTPUT_BYTES, runner::OutputBytesCallback};
4
5/// Persistent process readiness policy.
6#[derive(Debug, Clone, Eq, PartialEq)]
7#[non_exhaustive]
8pub enum PersistentReadiness {
9    /// The process is ready immediately after it is spawned.
10    Started,
11    /// The process is ready when either output stream contains the text.
12    OutputContains(String),
13    /// The process is ready when a command exits successfully.
14    Command(ProcessSpec),
15}
16
17/// Configuration for a persistent process.
18#[derive(Debug, Clone)]
19pub struct PersistentConfig {
20    /// Readiness policy used before returning the running process.
21    pub readiness: PersistentReadiness,
22    /// Maximum time to wait for readiness.
23    pub readiness_timeout: Duration,
24    /// Maximum time to wait after a graceful shutdown request before killing.
25    pub shutdown_grace_period: Duration,
26    /// Maximum retained bytes for each captured output stream.
27    pub max_capture_bytes: Option<usize>,
28    /// Output forwarding policy.
29    pub output: PersistentOutput,
30    /// Optional byte observers for every persistent process output chunk read.
31    pub output_observer: Option<PersistentOutputObserver>,
32}
33
34impl Default for PersistentConfig {
35    fn default() -> Self {
36        Self {
37            readiness: PersistentReadiness::Started,
38            readiness_timeout: Duration::from_secs(30),
39            shutdown_grace_period: Duration::from_secs(5),
40            max_capture_bytes: Some(DEFAULT_MAX_OUTPUT_BYTES),
41            output: PersistentOutput::capture_only(),
42            output_observer: None,
43        }
44    }
45}
46
47impl PersistentConfig {
48    /// Set the readiness policy.
49    #[must_use]
50    pub fn with_readiness(mut self, readiness: PersistentReadiness) -> Self {
51        self.readiness = readiness;
52        self
53    }
54
55    /// Set the readiness timeout.
56    #[must_use]
57    pub fn with_readiness_timeout(mut self, timeout: Duration) -> Self {
58        self.readiness_timeout = timeout;
59        self
60    }
61
62    /// Set the shutdown grace period.
63    #[must_use]
64    pub fn with_shutdown_grace_period(mut self, grace_period: Duration) -> Self {
65        self.shutdown_grace_period = grace_period;
66        self
67    }
68
69    /// Set the maximum retained bytes for each captured output stream.
70    #[must_use]
71    pub fn with_max_capture_bytes(mut self, bytes: usize) -> Self {
72        self.max_capture_bytes = Some(bytes);
73        self
74    }
75
76    /// Disable capture bounds.
77    #[must_use]
78    pub fn with_unbounded_capture(mut self) -> Self {
79        self.max_capture_bytes = None;
80        self
81    }
82
83    /// Set the output forwarding policy.
84    #[must_use]
85    pub fn with_output(mut self, output: PersistentOutput) -> Self {
86        self.output = output;
87        self
88    }
89
90    /// Set byte observers for persistent output.
91    #[must_use]
92    pub fn with_output_observer(mut self, observer: PersistentOutputObserver) -> Self {
93        self.output_observer = Some(observer);
94        self
95    }
96}
97
98/// Persistent process output forwarding policy.
99#[derive(Debug, Clone, Copy)]
100pub struct PersistentOutput {
101    stdout: Option<PersistentOutputStream>,
102    stderr: Option<PersistentOutputStream>,
103}
104
105impl PersistentOutput {
106    /// Capture output without forwarding it to the parent process streams.
107    #[must_use]
108    pub const fn capture_only() -> Self {
109        Self {
110            stdout: None,
111            stderr: None,
112        }
113    }
114
115    /// Capture and forward stdout/stderr to the selected parent streams.
116    #[must_use]
117    pub const fn forward(stdout: PersistentOutputStream, stderr: PersistentOutputStream) -> Self {
118        Self {
119            stdout: Some(stdout),
120            stderr: Some(stderr),
121        }
122    }
123
124    pub(in crate::persistent) const fn stdout_stream(&self) -> Option<PersistentOutputStream> {
125        self.stdout
126    }
127
128    pub(in crate::persistent) const fn stderr_stream(&self) -> Option<PersistentOutputStream> {
129        self.stderr
130    }
131}
132
133/// Byte observers for persistent process output.
134///
135/// Callbacks run on stdout/stderr reader threads with arbitrary chunk boundaries. Keep them
136/// fast and non-blocking; offload synchronous I/O or heavier processing to another worker to avoid
137/// stalling the reader thread and backing up the child process pipe.
138#[derive(Clone, Default)]
139pub struct PersistentOutputObserver {
140    stdout_bytes: Option<OutputBytesCallback>,
141    stderr_bytes: Option<OutputBytesCallback>,
142}
143
144impl std::fmt::Debug for PersistentOutputObserver {
145    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
146        formatter
147            .debug_struct("PersistentOutputObserver")
148            .field("stdout_bytes", &self.stdout_bytes.is_some())
149            .field("stderr_bytes", &self.stderr_bytes.is_some())
150            .finish()
151    }
152}
153
154impl PersistentOutputObserver {
155    /// Create a persistent output observer without callbacks.
156    #[must_use]
157    pub fn new() -> Self {
158        Self::default()
159    }
160
161    /// Observe each stdout byte chunk read from the persistent process.
162    ///
163    /// The callback runs on the stdout reader thread and must be fast/non-blocking.
164    #[must_use]
165    pub fn with_stdout_bytes(mut self, callback: impl Fn(&[u8]) + Send + Sync + 'static) -> Self {
166        self.stdout_bytes = Some(std::sync::Arc::new(callback));
167        self
168    }
169
170    /// Observe each stderr byte chunk read from the persistent process.
171    ///
172    /// The callback runs on the stderr reader thread and must be fast/non-blocking.
173    #[must_use]
174    pub fn with_stderr_bytes(mut self, callback: impl Fn(&[u8]) + Send + Sync + 'static) -> Self {
175        self.stderr_bytes = Some(std::sync::Arc::new(callback));
176        self
177    }
178
179    pub(in crate::persistent) fn stdout_bytes_callback(&self) -> Option<OutputBytesCallback> {
180        self.stdout_bytes.clone()
181    }
182
183    pub(in crate::persistent) fn stderr_bytes_callback(&self) -> Option<OutputBytesCallback> {
184        self.stderr_bytes.clone()
185    }
186}
187
188/// Parent stream used for forwarding persistent output.
189#[derive(Debug, Clone, Copy, Eq, PartialEq)]
190#[non_exhaustive]
191pub enum PersistentOutputStream {
192    /// Forward bytes to parent stdout.
193    Stdout,
194    /// Forward bytes to parent stderr.
195    Stderr,
196}
197
198#[cfg(test)]
199mod tests {
200    use super::*;
201
202    #[test]
203    fn persistent_config_builders_update_nested_policy() {
204        let readiness = PersistentReadiness::OutputContains("ready".to_string());
205        let output = PersistentOutput::forward(
206            PersistentOutputStream::Stdout,
207            PersistentOutputStream::Stderr,
208        );
209        let config = PersistentConfig::default()
210            .with_readiness(readiness.clone())
211            .with_readiness_timeout(Duration::from_secs(2))
212            .with_shutdown_grace_period(Duration::from_secs(3))
213            .with_max_capture_bytes(64)
214            .with_output(output);
215
216        assert_eq!(config.readiness, readiness);
217        assert_eq!(config.readiness_timeout, Duration::from_secs(2));
218        assert_eq!(config.shutdown_grace_period, Duration::from_secs(3));
219        assert_eq!(config.max_capture_bytes, Some(64));
220        assert_eq!(
221            config.output.stdout_stream(),
222            Some(PersistentOutputStream::Stdout)
223        );
224        assert_eq!(
225            config.output.stderr_stream(),
226            Some(PersistentOutputStream::Stderr)
227        );
228        assert!(config.output_observer.is_none());
229
230        assert_eq!(config.with_unbounded_capture().max_capture_bytes, None);
231    }
232
233    #[test]
234    fn persistent_output_remains_copy_forwarding_policy() {
235        let output = PersistentOutput::forward(
236            PersistentOutputStream::Stdout,
237            PersistentOutputStream::Stderr,
238        );
239        let copied = output;
240
241        assert_eq!(copied.stdout_stream(), Some(PersistentOutputStream::Stdout));
242        assert_eq!(output.stderr_stream(), Some(PersistentOutputStream::Stderr));
243    }
244
245    #[test]
246    fn persistent_output_observer_stores_byte_callbacks() {
247        let observer = PersistentOutputObserver::new()
248            .with_stdout_bytes(|_| {})
249            .with_stderr_bytes(|_| {});
250        let config = PersistentConfig::default().with_output_observer(observer);
251
252        let observer = config.output_observer.as_ref().expect("observer is set");
253        assert!(observer.stdout_bytes_callback().is_some());
254        assert!(observer.stderr_bytes_callback().is_some());
255    }
256}
rskit_process/persistent/config.rs

rskit_process/persistent/
config.rs
