tui_lipan/
process.rs

1//! Native-only helpers for streaming child-process output into components.
2//!
3//! This module is available only on non-wasm targets. It is intended for
4//! non-interactive subprocesses whose stdout/stderr should be consumed by the
5//! TUI while it keeps running. Interactive programs that need the real terminal
6//! should use [`crate::terminal_handoff`] instead.
7
8use std::io::{self, Read, Write};
9use std::path::{Path, PathBuf};
10use std::process::{Command as StdCommand, ExitStatus, Stdio};
11use std::sync::Arc;
12use std::sync::mpsc;
13use std::thread;
14use std::time::Duration;
15
16use crate::{Command, TaskPolicy};
17
18/// Description of a native child process to run and stream.
19#[derive(Clone, Debug, PartialEq, Eq)]
20pub struct ProcessSpec {
21    program: Arc<str>,
22    args: Vec<Arc<str>>,
23    cwd: Option<PathBuf>,
24    env: Vec<(Arc<str>, Arc<str>)>,
25    stdin: Option<Vec<u8>>,
26}
27
28impl ProcessSpec {
29    /// Create a process spec for `program`.
30    pub fn new(program: impl Into<Arc<str>>) -> Self {
31        Self {
32            program: program.into(),
33            args: Vec::new(),
34            cwd: None,
35            env: Vec::new(),
36            stdin: None,
37        }
38    }
39
40    /// Append a single argument.
41    #[must_use]
42    pub fn arg(mut self, arg: impl Into<Arc<str>>) -> Self {
43        self.args.push(arg.into());
44        self
45    }
46
47    /// Append multiple arguments.
48    #[must_use]
49    pub fn args<I, S>(mut self, args: I) -> Self
50    where
51        I: IntoIterator<Item = S>,
52        S: Into<Arc<str>>,
53    {
54        self.args.extend(args.into_iter().map(Into::into));
55        self
56    }
57
58    /// Set the child process working directory.
59    #[must_use]
60    pub fn cwd(mut self, cwd: impl Into<PathBuf>) -> Self {
61        self.cwd = Some(cwd.into());
62        self
63    }
64
65    /// Set an environment variable for the child process.
66    #[must_use]
67    pub fn env(mut self, key: impl Into<Arc<str>>, value: impl Into<Arc<str>>) -> Self {
68        self.env.push((key.into(), value.into()));
69        self
70    }
71
72    /// Provide bytes to write to the child process stdin, then close stdin.
73    #[must_use]
74    pub fn stdin(mut self, stdin: impl Into<Vec<u8>>) -> Self {
75        self.stdin = Some(stdin.into());
76        self
77    }
78
79    /// Program executable name or path.
80    pub fn program(&self) -> &str {
81        &self.program
82    }
83
84    /// Process arguments.
85    pub fn args_slice(&self) -> &[Arc<str>] {
86        &self.args
87    }
88
89    /// Working directory, when configured.
90    pub fn cwd_path(&self) -> Option<&Path> {
91        self.cwd.as_deref()
92    }
93
94    /// Environment overrides.
95    pub fn env_slice(&self) -> &[(Arc<str>, Arc<str>)] {
96        &self.env
97    }
98
99    /// Bytes that will be written to stdin, when configured.
100    pub fn stdin_bytes(&self) -> Option<&[u8]> {
101        self.stdin.as_deref()
102    }
103
104    /// Run the process on the current thread and emit streaming events.
105    ///
106    /// Stdout and stderr are drained on separate helper threads so a child that
107    /// writes heavily to both streams cannot deadlock on a full pipe.
108    pub fn stream(self, emit: impl FnMut(ProcessEvent)) -> io::Result<()> {
109        stream_process(self, emit)
110    }
111
112    /// Create an unkeyed background [`Command`] that streams process events as messages.
113    ///
114    /// Unkeyed commands do not have a normal runtime coalescing/cancellation
115    /// owner. Use [`Self::command_keyed`] when newer work should cancel an
116    /// active process.
117    pub fn command<Msg, F>(self, map: F) -> Command
118    where
119        Msg: Send + 'static,
120        F: Fn(ProcessEvent) -> Msg + Send + 'static,
121    {
122        process_command(self, map)
123    }
124
125    /// Create a keyed background [`Command`] that streams process events as messages.
126    ///
127    /// With [`TaskPolicy::LatestOnly`], submitting newer work for the same key
128    /// cancels the active token. This helper observes that token, kills the
129    /// child process, drains stdout/stderr, and suppresses stale messages.
130    pub fn command_keyed<Msg, F>(
131        self,
132        key: impl Into<Arc<str>>,
133        policy: TaskPolicy,
134        map: F,
135    ) -> Command
136    where
137        Msg: Send + 'static,
138        F: Fn(ProcessEvent) -> Msg + Send + 'static,
139    {
140        process_command_keyed(key, policy, self, map)
141    }
142
143    fn to_std_command(&self) -> StdCommand {
144        let mut command = StdCommand::new(self.program.as_ref());
145        command.args(self.args.iter().map(AsRef::<str>::as_ref));
146        if let Some(cwd) = &self.cwd {
147            command.current_dir(cwd);
148        }
149        for (key, value) in &self.env {
150            command.env(key.as_ref(), value.as_ref());
151        }
152        command.stdin(if self.stdin.is_some() {
153            Stdio::piped()
154        } else {
155            Stdio::null()
156        });
157        command.stdout(Stdio::piped());
158        command.stderr(Stdio::piped());
159        command
160    }
161}
162
163/// Framework-owned child-process exit status.
164#[derive(Clone, Copy, Debug, PartialEq, Eq)]
165pub struct ProcessExitStatus {
166    code: Option<i32>,
167    success: bool,
168}
169
170impl ProcessExitStatus {
171    /// Platform exit code, when the platform reports one.
172    pub fn code(self) -> Option<i32> {
173        self.code
174    }
175
176    /// Whether the process exited successfully.
177    pub fn success(self) -> bool {
178        self.success
179    }
180}
181
182impl From<ExitStatus> for ProcessExitStatus {
183    fn from(status: ExitStatus) -> Self {
184        Self {
185            code: status.code(),
186            success: status.success(),
187        }
188    }
189}
190
191/// Streaming child-process event.
192#[derive(Clone, Debug, PartialEq, Eq)]
193pub enum ProcessEvent {
194    /// Bytes read from stdout.
195    Stdout(Vec<u8>),
196    /// Bytes read from stderr.
197    Stderr(Vec<u8>),
198    /// Process exited and all captured output has been drained.
199    Exited(ProcessExitStatus),
200    /// Process-level or pipe-level error.
201    Error(Arc<str>),
202}
203
204/// Run a process on the current thread and emit stdout/stderr/exit events.
205pub fn stream_process(spec: ProcessSpec, mut emit: impl FnMut(ProcessEvent)) -> io::Result<()> {
206    stream_process_until(spec, || false, &mut emit)
207}
208
209/// Run a process and stop it when `should_cancel` returns `true`.
210///
211/// On cancellation the child is killed, stdout/stderr are drained, and an
212/// `Exited` event is emitted with the platform status reported by `wait`.
213pub fn stream_process_until(
214    mut spec: ProcessSpec,
215    should_cancel: impl Fn() -> bool,
216    mut emit: impl FnMut(ProcessEvent),
217) -> io::Result<()> {
218    let mut command = spec.to_std_command();
219    let stdin_bytes = spec.stdin.take();
220    let mut child = command.spawn()?;
221
222    let stdout = child.stdout.take();
223    let stderr = child.stderr.take();
224    let child_stdin = child.stdin.take();
225    let (tx, rx) = mpsc::channel();
226
227    let stdout_thread = stdout.map(|stdout| spawn_reader(stdout, tx.clone(), ProcessEvent::Stdout));
228    let stderr_thread = stderr.map(|stderr| spawn_reader(stderr, tx.clone(), ProcessEvent::Stderr));
229    let stdin_thread = stdin_bytes.and_then(|bytes| {
230        child_stdin.map(|mut stdin| {
231            let tx = tx.clone();
232            thread::spawn(move || {
233                if let Err(err) = stdin.write_all(&bytes) {
234                    send_error(&tx, err);
235                }
236            })
237        })
238    });
239
240    let status = loop {
241        while let Ok(event) = rx.try_recv() {
242            emit(event);
243        }
244
245        if should_cancel() {
246            let _ = child.kill();
247            break child.wait();
248        }
249
250        if let Some(status) = child.try_wait()? {
251            break Ok(status);
252        }
253
254        match rx.recv_timeout(Duration::from_millis(20)) {
255            Ok(event) => emit(event),
256            Err(mpsc::RecvTimeoutError::Timeout) => {}
257            Err(mpsc::RecvTimeoutError::Disconnected) => {
258                if let Some(status) = child.try_wait()? {
259                    break Ok(status);
260                }
261            }
262        }
263    };
264
265    join_optional(stdin_thread);
266    join_optional(stdout_thread);
267    join_optional(stderr_thread);
268
269    for event in rx.try_iter() {
270        emit(event);
271    }
272
273    emit(ProcessEvent::Exited(status?.into()));
274    Ok(())
275}
276
277/// Create a background command that maps process events into component messages.
278pub fn process_command<Msg, F>(spec: ProcessSpec, map: F) -> Command
279where
280    Msg: Send + 'static,
281    F: Fn(ProcessEvent) -> Msg + Send + 'static,
282{
283    Command::spawn::<Msg, _>(move |link| {
284        if let Err(err) = stream_process_until(
285            spec,
286            || link.is_cancelled(),
287            |event| {
288                let _ = link.send_if_not_cancelled(map(event));
289            },
290        ) {
291            let _ =
292                link.send_if_not_cancelled(map(ProcessEvent::Error(Arc::from(err.to_string()))));
293        }
294    })
295}
296
297/// Create a keyed background command that maps process events into component messages.
298pub fn process_command_keyed<Msg, F>(
299    key: impl Into<Arc<str>>,
300    policy: TaskPolicy,
301    spec: ProcessSpec,
302    map: F,
303) -> Command
304where
305    Msg: Send + 'static,
306    F: Fn(ProcessEvent) -> Msg + Send + 'static,
307{
308    Command::spawn_keyed::<Msg, _>(key, policy, move |link| {
309        if let Err(err) = stream_process_until(
310            spec,
311            || link.is_cancelled(),
312            |event| {
313                let _ = link.send_if_not_cancelled(map(event));
314            },
315        ) {
316            let _ =
317                link.send_if_not_cancelled(map(ProcessEvent::Error(Arc::from(err.to_string()))));
318        }
319    })
320}
321
322fn spawn_reader<R>(
323    mut reader: R,
324    tx: mpsc::Sender<ProcessEvent>,
325    wrap: fn(Vec<u8>) -> ProcessEvent,
326) -> thread::JoinHandle<()>
327where
328    R: Read + Send + 'static,
329{
330    thread::spawn(move || {
331        let mut buf = [0_u8; 8192];
332        loop {
333            match reader.read(&mut buf) {
334                Ok(0) => break,
335                Ok(n) => {
336                    if tx.send(wrap(buf[..n].to_vec())).is_err() {
337                        break;
338                    }
339                }
340                Err(err) => {
341                    send_error(&tx, err);
342                    break;
343                }
344            }
345        }
346    })
347}
348
349fn join_optional(handle: Option<thread::JoinHandle<()>>) {
350    if let Some(handle) = handle {
351        let _ = handle.join();
352    }
353}
354
355fn send_error(tx: &mpsc::Sender<ProcessEvent>, err: io::Error) {
356    let _ = tx.send(ProcessEvent::Error(Arc::from(err.to_string())));
357}
358
359#[cfg(test)]
360mod tests {
361    use super::*;
362
363    #[test]
364    fn process_spec_builder_sets_fields() {
365        let spec = ProcessSpec::new("sh")
366            .arg("-c")
367            .args(["printf", "ok"])
368            .cwd("/")
369            .env("KEY", "VALUE")
370            .stdin(b"input".to_vec());
371
372        assert_eq!(spec.program(), "sh");
373        assert_eq!(spec.args_slice().len(), 3);
374        assert_eq!(spec.args_slice()[0].as_ref(), "-c");
375        assert_eq!(spec.cwd_path(), Some(Path::new("/")));
376        assert_eq!(spec.env_slice()[0].0.as_ref(), "KEY");
377        assert_eq!(spec.env_slice()[0].1.as_ref(), "VALUE");
378        assert_eq!(spec.stdin_bytes(), Some(b"input".as_slice()));
379    }
380
381    #[cfg(unix)]
382    #[test]
383    fn streams_stdout_stderr_and_exit() {
384        let spec = ProcessSpec::new("sh").args(["-c", "printf out; printf err >&2"]);
385        let mut events = Vec::new();
386
387        spec.stream(|event| events.push(event)).unwrap();
388
389        let stdout: Vec<u8> = events
390            .iter()
391            .filter_map(|event| match event {
392                ProcessEvent::Stdout(bytes) => Some(bytes.as_slice()),
393                _ => None,
394            })
395            .flatten()
396            .copied()
397            .collect();
398        let stderr: Vec<u8> = events
399            .iter()
400            .filter_map(|event| match event {
401                ProcessEvent::Stderr(bytes) => Some(bytes.as_slice()),
402                _ => None,
403            })
404            .flatten()
405            .copied()
406            .collect();
407
408        assert_eq!(stdout, b"out");
409        assert_eq!(stderr, b"err");
410        assert!(matches!(events.last(), Some(ProcessEvent::Exited(status)) if status.success()));
411    }
412
413    #[cfg(unix)]
414    #[test]
415    fn stream_process_until_kills_child_when_cancelled() {
416        let spec = ProcessSpec::new("sh").args(["-c", "sleep 5"]);
417        let mut events = Vec::new();
418
419        stream_process_until(spec, || true, |event| events.push(event)).unwrap();
420
421        assert!(matches!(events.last(), Some(ProcessEvent::Exited(status)) if !status.success()));
422    }
423}
tui_lipan/process.rs

tui_lipan/
process.rs
