//! Drives a headless `pixelpass` child process for the GUI. //! //! The GUI re-execs this same binary (via [`std::env::current_exe`]) in //! headless mode with `--output json`, then reads the child's JSON event //! stream on a background thread and forwards parsed events over a channel the //! egui app drains each frame. stderr is captured into a small ring so a //! failed launch (e.g. a missing gst plugin) can be surfaced in the window. use std::io::{BufRead, BufReader, Write}; use std::process::{Child, ChildStdin, Command, Stdio}; use std::sync::mpsc::Receiver; use std::sync::{Arc, Mutex}; use std::time::Duration; use nix::sys::signal::{Signal, kill}; use nix::unistd::Pid; use serde::Deserialize; use super::Waker; /// One parsed event from the child's stdout. Owned mirror of /// [`crate::common::output::Event`] (which borrows for emit); kept separate so /// the wire format and the parser can evolve independently. #[derive(Deserialize, Debug)] #[serde(tag = "event", rename_all = "snake_case")] pub enum ChildEvent { Ticket { value: String, }, HostInfo { display_server: String, capture: String, quality: String, dimensions: String, hw_encode: bool, max_viewers: u32, max_viewers_source: String, }, ViewerJoined { id: String, active: u32, max: u32, }, ViewerLeft { id: String, active: u32, max: u32, }, Capture { state: CaptureState, }, ViewerRefused { reason: String, }, Connected { url: String, }, } #[derive(Deserialize, Debug, Clone, Copy, PartialEq)] #[serde(rename_all = "snake_case")] pub enum CaptureState { Started, Stopped, } const STDERR_TAIL_MAX: usize = 60; pub struct ChildProc { /// `Some` while the child is owned here; `Drop` takes it to hand off to a /// detached reaper thread (see the `Drop` impl). child: Option, pub rx: Receiver, stderr_tail: Arc>>, /// Write end of the child's stdin, for the line-based command channel /// (see [`ChildProc::send_command`]). `None` once it's been closed. stdin: Option, } impl ChildProc { /// Spawn `pixelpass ` as a child, wiring up the event reader. The /// `waker` is pinged whenever an event arrives so the UI thread wakes to /// drain it — this wakes the winit event loop directly (via an /// `EventLoopProxy`), so it works even when the window is hidden to the tray /// and no frames are running (egui's own repaint callback would not fire /// repeatedly in that idle state — see [`super::Waker`]). pub fn spawn(args: &[String], waker: Waker) -> std::io::Result { let exe = std::env::current_exe()?; let mut child = Command::new(exe) .args(args) // Piped so we can send line commands (e.g. `kick `); the host // only reads it when driven this way (`--output json`). .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn()?; let stdin = child.stdin.take(); let (tx, rx) = std::sync::mpsc::channel(); let stdout = child.stdout.take().expect("stdout piped"); std::thread::spawn(move || { let reader = BufReader::new(stdout); for line in reader.lines().map_while(Result::ok) { if line.trim().is_empty() { continue; } // Ignore any non-event line rather than dropping the stream. if let Ok(ev) = serde_json::from_str::(&line) { if tx.send(ev).is_err() { break; // app gone } waker.wake(); } } // stdout closed → the child has exited (player closed, connection // ended, or a failed launch). Wake once more so the UI reaps it and // clears the "running" view, even if no final event was emitted and // the window is hidden to the tray. waker.wake(); }); let stderr_tail = Arc::new(Mutex::new(Vec::::new())); let stderr = child.stderr.take().expect("stderr piped"); let tail = stderr_tail.clone(); std::thread::spawn(move || { let reader = BufReader::new(stderr); for line in reader.lines().map_while(Result::ok) { let mut t = tail.lock().unwrap(); t.push(line); let overflow = t.len().saturating_sub(STDERR_TAIL_MAX); if overflow > 0 { t.drain(0..overflow); } } }); Ok(Self { child: Some(child), rx, stderr_tail, stdin, }) } /// Send one newline-terminated command to the child over its stdin (the /// host parses these as `kick `). Best-effort: a closed pipe /// (child already gone) just drops the command. pub fn send_command(&mut self, cmd: &str) { let Some(stdin) = self.stdin.as_mut() else { return; }; if let Err(e) = writeln!(stdin, "{cmd}") { tracing::warn!("failed to send command to host child: {e}"); self.stdin = None; // pipe is dead; stop trying } } /// Whether the child is still running. pub fn is_alive(&mut self) -> bool { matches!(self.child.as_mut().map(Child::try_wait), Some(Ok(None))) } /// The last captured stderr lines, joined — for error display. pub fn stderr_tail(&self) -> String { self.stderr_tail.lock().unwrap().join("\n") } } impl Drop for ChildProc { fn drop(&mut self) { // Leaving a host/viewer screen, or closing the window, must not orphan // a live child — but it must also not *block*. eframe runs this drop // synchronously while it destroys the window, so a grace-period wait // here freezes the window mid-close: the first click looks like it did // nothing (the stream just drops) and the window only goes away on a // second click. So SIGINT now — synchronously, so the host always gets // its ctrl-c teardown (capture down, endpoint closed) even if we exit // right after — then reap on a detached thread instead of waiting. let Some(mut child) = self.child.take() else { return; }; if matches!(child.try_wait(), Ok(Some(_))) { return; // already exited; nothing to signal or reap } let _ = kill(Pid::from_raw(child.id() as i32), Signal::SIGINT); std::thread::spawn(move || { for _ in 0..40 { if matches!(child.try_wait(), Ok(Some(_))) { return; } std::thread::sleep(Duration::from_millis(50)); } let _ = child.kill(); let _ = child.wait(); }); } } #[cfg(test)] mod tests { use super::*; use crate::common::output::{CaptureState as EmitState, Event}; // The GUI parses what the headless child emits. These round-trip the // emitter's own types (`common::output::Event`) through the parser // (`ChildEvent`) so a rename on either side of the wire fails here rather // than silently breaking the GUI at runtime. fn parse(emit: Event) -> ChildEvent { let line = serde_json::to_string(&emit).unwrap(); serde_json::from_str::(&line).unwrap() } #[test] fn ticket_round_trips() { assert!(matches!( parse(Event::Ticket { value: "endpointXYZ" }), ChildEvent::Ticket { value } if value == "endpointXYZ" )); } #[test] fn host_info_round_trips() { let ev = parse(Event::HostInfo { display_server: "Wayland", capture: "fullscreen + system-audio", quality: "Medium", dimensions: "≤720p / 2500 kbps / 30 fps", hw_encode: true, max_viewers: 3, max_viewers_source: "user-specified", }); match ev { ChildEvent::HostInfo { display_server, quality, hw_encode, max_viewers, .. } => { assert_eq!(display_server, "Wayland"); assert_eq!(quality, "Medium"); assert!(hw_encode); assert_eq!(max_viewers, 3); } other => panic!("expected HostInfo, got {other:?}"), } } #[test] fn viewer_join_leave_round_trip() { assert!(matches!( parse(Event::ViewerJoined { id: "nodeXYZ", active: 2, max: 4 }), ChildEvent::ViewerJoined { id, active: 2, max: 4 } if id == "nodeXYZ" )); assert!(matches!( parse(Event::ViewerLeft { id: "nodeXYZ", active: 1, max: 4 }), ChildEvent::ViewerLeft { id, active: 1, max: 4 } if id == "nodeXYZ" )); } #[test] fn capture_state_round_trips() { assert!(matches!( parse(Event::Capture { state: EmitState::Started }), ChildEvent::Capture { state: CaptureState::Started } )); assert!(matches!( parse(Event::Capture { state: EmitState::Stopped }), ChildEvent::Capture { state: CaptureState::Stopped } )); } #[test] fn refused_and_connected_round_trip() { assert!(matches!( parse(Event::ViewerRefused { reason: "host is full" }), ChildEvent::ViewerRefused { reason } if reason == "host is full" )); assert!(matches!( parse(Event::Connected { url: "http://127.0.0.1:5000" }), ChildEvent::Connected { url } if url == "http://127.0.0.1:5000" )); } }