hexomc_lib/launch/
output.rs

1use std::{process::ExitStatus, sync::Arc};
2
3use tokio::{
4    io::{AsyncBufReadExt, AsyncRead, BufReader},
5    task::JoinHandle,
6};
7
8use crate::error::Result;
9
10/// Which stream a line came from.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub enum OutputKind {
13    Stdout,
14    Stderr,
15}
16
17/// One line of game output, with its source stream.
18#[derive(Debug, Clone)]
19pub struct OutputLine {
20    pub kind: OutputKind,
21    pub line: String,
22}
23
24impl OutputLine {
25    pub fn is_stderr(&self) -> bool {
26        self.kind == OutputKind::Stderr
27    }
28}
29
30/// Output callback. `Arc` so it can be cloned into the two reader tasks
31/// (stdout / stderr) and outlive the caller's stack frame.
32///
33/// The callback runs on a tokio task, so it must not block for long; push the
34/// line into a channel / log if any real work is needed.
35pub type OutputFn = Arc<dyn Fn(OutputLine) + Send + Sync + 'static>;
36
37pub fn no_output() -> OutputFn {
38    Arc::new(|_| {})
39}
40
41/// A running game process whose stdout/stderr are being piped to an `OutputFn`.
42///
43/// Dropping this does *not* kill the game (same as `std::process::Child`); the
44/// reader tasks end on their own when the pipes close.
45pub struct GameProcess {
46    child: tokio::process::Child,
47    /// Cached: `Child::id()` returns None once the process has been waited on.
48    pid: Option<u32>,
49    readers: Vec<JoinHandle<()>>,
50}
51
52impl GameProcess {
53    /// Take the piped stdout/stderr and start pumping lines into `sink`.
54    pub(crate) fn new(mut child: tokio::process::Child, sink: OutputFn) -> Self {
55        let pid = child.id();
56        let mut readers = Vec::new();
57
58        if let Some(stdout) = child.stdout.take() {
59            readers.push(spawn_pump(stdout, OutputKind::Stdout, sink.clone()));
60        }
61        if let Some(stderr) = child.stderr.take() {
62            readers.push(spawn_pump(stderr, OutputKind::Stderr, sink));
63        }
64
65        Self { child, pid, readers }
66    }
67
68    /// PID of the game process (None once it has exited).
69    pub fn id(&self) -> Option<u32> {
70        self.pid
71    }
72
73    /// Wait for the game to exit, then for every pending output line to be
74    /// delivered — after this returns, the callback will not fire again.
75    pub async fn wait(&mut self) -> Result<ExitStatus> {
76        let status = self.child.wait().await?;
77        for reader in self.readers.drain(..) {
78            let _ = reader.await;
79        }
80        Ok(status)
81    }
82
83    /// Check whether the game has exited, without blocking.
84    pub fn try_wait(&mut self) -> Result<Option<ExitStatus>> {
85        Ok(self.child.try_wait()?)
86    }
87
88    /// Kill the game and reap it.
89    pub async fn kill(&mut self) -> Result<()> {
90        self.child.kill().await?;
91        Ok(())
92    }
93
94    /// Escape hatch for anything not covered above (e.g. `start_kill`).
95    pub fn child_mut(&mut self) -> &mut tokio::process::Child {
96        &mut self.child
97    }
98}
99
100/// Read `reader` line by line into `sink` until EOF.
101///
102/// Uses `read_until` + `from_utf8_lossy` rather than `lines()`: crash reports and
103/// some mods emit non-UTF-8 bytes, which would abort a `lines()` stream.
104fn spawn_pump<R>(reader: R, kind: OutputKind, sink: OutputFn) -> JoinHandle<()>
105where
106    R: AsyncRead + Unpin + Send + 'static,
107{
108    tokio::spawn(async move {
109        let mut buf = BufReader::new(reader);
110        let mut bytes = Vec::new();
111
112        loop {
113            bytes.clear();
114            match buf.read_until(b'\n', &mut bytes).await {
115                Ok(0) | Err(_) => break,
116                Ok(_) => {
117                    while matches!(bytes.last(), Some(b'\n') | Some(b'\r')) {
118                        bytes.pop();
119                    }
120                    let line = String::from_utf8_lossy(&bytes).into_owned();
121                    sink(OutputLine { kind, line });
122                }
123            }
124        }
125    })
126}
127
128#[cfg(test)]
129mod tests {
130    use super::*;
131    use std::sync::Mutex;
132
133    #[tokio::test]
134    async fn pump_splits_lines_and_strips_crlf() {
135        let data: &[u8] = b"first\r\nsecond\nthird-no-newline";
136        let collected = Arc::new(Mutex::new(Vec::new()));
137        let sink_store = collected.clone();
138        let sink: OutputFn = Arc::new(move |l: OutputLine| {
139            sink_store.lock().unwrap().push(l.line);
140        });
141
142        spawn_pump(data, OutputKind::Stdout, sink).await.unwrap();
143
144        let lines = collected.lock().unwrap().clone();
145        assert_eq!(lines, vec!["first", "second", "third-no-newline"]);
146    }
147
148    #[tokio::test]
149    async fn pump_survives_invalid_utf8() {
150        let data: &[u8] = b"ok\n\xff\xfe bad\n";
151        let collected = Arc::new(Mutex::new(Vec::new()));
152        let sink_store = collected.clone();
153        let sink: OutputFn = Arc::new(move |l: OutputLine| {
154            sink_store.lock().unwrap().push(l.line);
155        });
156
157        spawn_pump(data, OutputKind::Stderr, sink).await.unwrap();
158
159        let lines = collected.lock().unwrap().clone();
160        assert_eq!(lines.len(), 2);
161        assert_eq!(lines[0], "ok");
162        assert!(lines[1].ends_with(" bad"));
163    }
164}
hexomc_lib/launch/output.rs

hexomc_lib/launch/
output.rs
