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 fast
136/// and non-blocking; offload synchronous I/O
137/// or heavier processing to another worker to avoid stalling the reader thread
138/// and backing up the child process pipe.
139#[derive(Clone, Default)]
140pub struct PersistentOutputObserver {
141    stdout_bytes: Option<OutputBytesCallback>,
142    stderr_bytes: Option<OutputBytesCallback>,
143}
144
145impl std::fmt::Debug for PersistentOutputObserver {
146    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
147        formatter
148            .debug_struct("PersistentOutputObserver")
149            .field("stdout_bytes", &self.stdout_bytes.is_some())
150            .field("stderr_bytes", &self.stderr_bytes.is_some())
151            .finish()
152    }
153}
154
155impl PersistentOutputObserver {
156    /// Create a persistent output observer without callbacks.
157    #[must_use]
158    pub fn new() -> Self {
159        Self::default()
160    }
161
162    /// Observe each stdout byte chunk read from the persistent process.
163    ///
164    /// The callback runs on the stdout reader thread and must be fast/non-blocking.
165    #[must_use]
166    pub fn with_stdout_bytes(mut self, callback: impl Fn(&[u8]) + Send + Sync + 'static) -> Self {
167        self.stdout_bytes = Some(std::sync::Arc::new(callback));
168        self
169    }
170
171    /// Observe each stderr byte chunk read from the persistent process.
172    ///
173    /// The callback runs on the stderr reader thread and must be fast/non-blocking.
174    #[must_use]
175    pub fn with_stderr_bytes(mut self, callback: impl Fn(&[u8]) + Send + Sync + 'static) -> Self {
176        self.stderr_bytes = Some(std::sync::Arc::new(callback));
177        self
178    }
179
180    pub(in crate::persistent) fn stdout_bytes_callback(&self) -> Option<OutputBytesCallback> {
181        self.stdout_bytes.clone()
182    }
183
184    pub(in crate::persistent) fn stderr_bytes_callback(&self) -> Option<OutputBytesCallback> {
185        self.stderr_bytes.clone()
186    }
187}
188
189/// Parent stream used for forwarding persistent output.
190#[derive(Debug, Clone, Copy, Eq, PartialEq)]
191#[non_exhaustive]
192pub enum PersistentOutputStream {
193    /// Forward bytes to parent stdout.
194    Stdout,
195    /// Forward bytes to parent stderr.
196    Stderr,
197}
198
199#[cfg(test)]
200mod tests {
201    use super::*;
202
203    #[test]
204    fn persistent_config_builders_update_nested_policy() {
205        let readiness = PersistentReadiness::OutputContains("ready".to_string());
206        let output = PersistentOutput::forward(
207            PersistentOutputStream::Stdout,
208            PersistentOutputStream::Stderr,
209        );
210        let config = PersistentConfig::default()
211            .with_readiness(readiness.clone())
212            .with_readiness_timeout(Duration::from_secs(2))
213            .with_shutdown_grace_period(Duration::from_secs(3))
214            .with_max_capture_bytes(64)
215            .with_output(output);
216
217        assert_eq!(config.readiness, readiness);
218        assert_eq!(config.readiness_timeout, Duration::from_secs(2));
219        assert_eq!(config.shutdown_grace_period, Duration::from_secs(3));
220        assert_eq!(config.max_capture_bytes, Some(64));
221        assert_eq!(
222            config.output.stdout_stream(),
223            Some(PersistentOutputStream::Stdout)
224        );
225        assert_eq!(
226            config.output.stderr_stream(),
227            Some(PersistentOutputStream::Stderr)
228        );
229        assert!(config.output_observer.is_none());
230
231        assert_eq!(config.with_unbounded_capture().max_capture_bytes, None);
232    }
233
234    #[test]
235    fn persistent_output_remains_copy_forwarding_policy() {
236        let output = PersistentOutput::forward(
237            PersistentOutputStream::Stdout,
238            PersistentOutputStream::Stderr,
239        );
240        let copied = output;
241
242        assert_eq!(copied.stdout_stream(), Some(PersistentOutputStream::Stdout));
243        assert_eq!(output.stderr_stream(), Some(PersistentOutputStream::Stderr));
244    }
245
246    #[test]
247    fn persistent_output_observer_stores_byte_callbacks() {
248        let observer = PersistentOutputObserver::new()
249            .with_stdout_bytes(|_| {})
250            .with_stderr_bytes(|_| {});
251        let config = PersistentConfig::default().with_output_observer(observer);
252
253        let observer = config.output_observer.as_ref().expect("observer is set");
254        assert!(observer.stdout_bytes_callback().is_some());
255        assert!(observer.stderr_bytes_callback().is_some());
256    }
257
258    #[test]
259    fn default_and_observer_debug_are_stable() {
260        let config = PersistentConfig::default();
261        assert_eq!(config.readiness, PersistentReadiness::Started);
262        assert_eq!(config.readiness_timeout, Duration::from_secs(30));
263        assert_eq!(config.shutdown_grace_period, Duration::from_secs(5));
264
265        let observer = PersistentOutputObserver::new().with_stdout_bytes(|_| {});
266        let rendered = format!("{observer:?}");
267        assert!(rendered.contains("stdout_bytes: true"));
268        assert!(rendered.contains("stderr_bytes: false"));
269    }
270}
rskit_process/persistent/config.rs

rskit_process/persistent/
config.rs
