windjammer_runtime/
subprocess.rs

1//! Interactive subprocess I/O (spawn + piped stdin/stdout).
2//!
3//! Windjammer `std::subprocess` maps here. Uses opaque handles so backends can swap
4//! implementation without changing the WJ API.
5
6use std::collections::{HashMap, VecDeque};
7use std::io::{BufRead, BufReader, Write};
8use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio};
9use std::sync::atomic::{AtomicU64, Ordering};
10use std::sync::{Arc, Condvar, Mutex};
11use std::thread::JoinHandle;
12
13use once_cell::sync::Lazy;
14
15static NEXT_HANDLE: AtomicU64 = AtomicU64::new(1);
16static SESSIONS: Lazy<Mutex<HashMap<u64, SubprocessInner>>> =
17    Lazy::new(|| Mutex::new(HashMap::new()));
18
19type LineQueue = Arc<(Mutex<VecDeque<String>>, Condvar)>;
20
21struct SubprocessInner {
22    child: Child,
23    stdin: ChildStdin,
24    lines: LineQueue,
25    _drainer: JoinHandle<()>,
26}
27
28/// Opaque handle returned to Windjammer (`Subprocess.handle`).
29#[derive(Debug, Clone, Copy, PartialEq, Eq)]
30pub struct SubprocessHandle {
31    pub id: u64,
32}
33
34impl SubprocessHandle {
35    pub fn invalid() -> Self {
36        Self { id: 0 }
37    }
38
39    pub fn is_valid(self) -> bool {
40        self.id != 0
41    }
42}
43
44pub fn invalid_handle() -> SubprocessHandle {
45    SubprocessHandle::invalid()
46}
47
48pub fn is_valid(handle: SubprocessHandle) -> bool {
49    handle.is_valid()
50}
51
52fn start_stdout_drainer(stdout: ChildStdout) -> (LineQueue, JoinHandle<()>) {
53    let lines: LineQueue = Arc::new((Mutex::new(VecDeque::new()), Condvar::new()));
54    let lines_clone = Arc::clone(&lines);
55    let drainer = std::thread::spawn(move || {
56        let mut reader = BufReader::new(stdout);
57        loop {
58            let mut buf = String::new();
59            match reader.read_line(&mut buf) {
60                Ok(0) => break,
61                Ok(_) => {
62                    let line = buf.trim_end().to_string();
63                    let (lock, cvar) = &*lines_clone;
64                    let mut guard = lock.lock().unwrap();
65                    guard.push_back(line);
66                    cvar.notify_one();
67                }
68                Err(_) => break,
69            }
70        }
71        let (lock, cvar) = &*lines_clone;
72        let mut guard = lock.lock().unwrap();
73        guard.push_back(String::new()); // EOF sentinel
74        cvar.notify_all();
75    });
76    (lines, drainer)
77}
78
79fn pop_line(lines: &LineQueue) -> Result<String, String> {
80    let (lock, cvar) = &**lines;
81    let mut guard = lock.lock().unwrap();
82    loop {
83        if let Some(line) = guard.pop_front() {
84            if line.is_empty() {
85                return Err("stdout closed".to_string());
86            }
87            return Ok(line);
88        }
89        guard = cvar.wait(guard).map_err(|e| e.to_string())?;
90    }
91}
92
93/// Spawn a program with piped stdin/stdout (stderr inherits to host).
94/// A background thread drains stdout so game init logs cannot deadlock the pipe.
95pub fn spawn(program: &str, args: &[String]) -> Result<SubprocessHandle, String> {
96    let mut child = Command::new(program)
97        .args(args)
98        .stdin(Stdio::piped())
99        .stdout(Stdio::piped())
100        .stderr(Stdio::inherit())
101        .spawn()
102        .map_err(|e| format!("spawn {}: {}", program, e))?;
103
104    let stdin = child
105        .stdin
106        .take()
107        .ok_or_else(|| "stdin unavailable".to_string())?;
108    let stdout = child
109        .stdout
110        .take()
111        .ok_or_else(|| "stdout unavailable".to_string())?;
112
113    let (lines, drainer) = start_stdout_drainer(stdout);
114
115    let id = NEXT_HANDLE.fetch_add(1, Ordering::SeqCst);
116    SESSIONS.lock().unwrap().insert(
117        id,
118        SubprocessInner {
119            child,
120            stdin,
121            lines,
122            _drainer: drainer,
123        },
124    );
125    Ok(SubprocessHandle { id })
126}
127
128fn with_session<F, T>(handle: SubprocessHandle, f: F) -> Result<T, String>
129where
130    F: FnOnce(&mut SubprocessInner) -> Result<T, String>,
131{
132    if !handle.is_valid() {
133        return Err("invalid subprocess handle".to_string());
134    }
135    let mut guard = SESSIONS.lock().unwrap();
136    let session = guard
137        .get_mut(&handle.id)
138        .ok_or_else(|| "subprocess not found".to_string())?;
139    f(session)
140}
141
142pub fn write_line(handle: SubprocessHandle, line: &str) -> Result<(), String> {
143    with_session(handle, |s| {
144        writeln!(s.stdin, "{}", line).map_err(|e| e.to_string())?;
145        s.stdin.flush().map_err(|e| e.to_string())
146    })
147}
148
149pub fn read_line(handle: SubprocessHandle) -> Result<String, String> {
150    with_session(handle, |s| pop_line(&s.lines))
151}
152
153/// Read lines until one starts with `prefix` (e.g. `@AGENT OBS`).
154pub fn read_line_until_prefix(handle: SubprocessHandle, prefix: &str) -> Result<String, String> {
155    loop {
156        let line = read_line(handle)?;
157        if line.starts_with(prefix) {
158            return Ok(line);
159        }
160    }
161}
162
163const AGENT_OBS_PREFIX: &str = "@AGENT OBS ";
164
165/// Parse `frames=N` from an agent CMD payload (default 1).
166pub fn frames_in_agent_command(command: &str) -> i64 {
167    for part in command.split(';') {
168        let part = part.trim();
169        if let Some(v) = part.strip_prefix("frames=") {
170            if let Ok(n) = v.parse::<i64>() {
171                return n.max(1);
172            }
173        }
174    }
175    1
176}
177
178/// Read N `@AGENT OBS` lines; returns the payload of the last one (no prefix).
179pub fn read_agent_obs_n(handle: SubprocessHandle, count: i64) -> Result<String, String> {
180    let n = count.max(1) as u32;
181    let mut last = String::new();
182    for _ in 0..n {
183        let line = read_line_until_prefix(handle, AGENT_OBS_PREFIX)?;
184        last = line
185            .strip_prefix(AGENT_OBS_PREFIX)
186            .unwrap_or(&line)
187            .trim()
188            .to_string();
189    }
190    Ok(last)
191}
192
193pub fn kill(handle: SubprocessHandle) -> Result<(), String> {
194    with_session(handle, |s| s.child.kill().map_err(|e| e.to_string()))
195}
196
197pub fn wait(handle: SubprocessHandle) -> Result<i32, String> {
198    with_session(handle, |s| {
199        s.child
200            .wait()
201            .map_err(|e| e.to_string())
202            .map(|st| st.code().unwrap_or(-1))
203    })
204}
205
206pub fn close(handle: SubprocessHandle) -> Result<(), String> {
207    if !handle.is_valid() {
208        return Ok(());
209    }
210    let mut guard = SESSIONS.lock().unwrap();
211    guard.remove(&handle.id);
212    Ok(())
213}
214
215#[cfg(test)]
216mod tests {
217    use super::*;
218
219    #[test]
220    fn spawn_echo_line() {
221        #[cfg(unix)]
222        {
223            let h = spawn("echo", &["hello".to_string()]).unwrap();
224            let line = read_line(h).unwrap();
225            assert_eq!(line.trim(), "hello");
226            let _ = wait(h);
227            let _ = close(h);
228        }
229    }
230
231    #[test]
232    fn spawn_cat_write_read() {
233        #[cfg(unix)]
234        {
235            let h = spawn("cat", &[]).unwrap();
236            write_line(h, "ping").unwrap();
237            let line = read_line(h).unwrap();
238            assert_eq!(line, "ping");
239            let _ = kill(h);
240            let _ = close(h);
241        }
242    }
243
244    #[test]
245    fn frames_in_agent_command_parses() {
246        assert_eq!(frames_in_agent_command("frames=30;forward=1"), 30);
247        assert_eq!(frames_in_agent_command("forward=1"), 1);
248    }
249}
windjammer_runtime/subprocess.rs

windjammer_runtime/
subprocess.rs
