feat(gui): host + viewer tabs driving the headless child
The GUI now does real work. Host tab: a config form (quality combo, max-viewers, software-encode + single-window toggles) spawns `pixelpass --host --output json …` via re-exec, then a background thread parses the child's JSON events and the window shows live status — ticket with a copy button, viewer count, streaming/waiting state, host_info summary, and host-full refusals. Viewer tab: paste a code, pick mpv/VLC, Connect spawns `pixelpass <ticket> --output json`, and on the connected event the GUI launches the player (reusing interactive::Player). ChildProc (gui/child.rs) owns the child: reads stdout events over a channel, rings the last 60 stderr lines for failure display, and stops via SIGINT (graceful host teardown) with a 2s grace before SIGKILL — Drop ensures closing the window never orphans a live host. Five round-trip tests lock the common::output::Event ↔ ChildEvent wire contract. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.7
@@ -0,0 +1,238 @@
|
||||
//! 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};
|
||||
use std::process::{Child, Command, Stdio};
|
||||
use std::sync::mpsc::Receiver;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use eframe::egui;
|
||||
use nix::sys::signal::{Signal, kill};
|
||||
use nix::unistd::Pid;
|
||||
use serde::Deserialize;
|
||||
|
||||
/// 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,
|
||||
},
|
||||
ViewerCount {
|
||||
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 {
|
||||
child: Child,
|
||||
pub rx: Receiver<ChildEvent>,
|
||||
stderr_tail: Arc<Mutex<Vec<String>>>,
|
||||
}
|
||||
|
||||
impl ChildProc {
|
||||
/// Spawn `pixelpass <args>` as a child, wiring up the event reader. `ctx`
|
||||
/// is repainted whenever an event arrives so the UI updates live.
|
||||
pub fn spawn(args: &[String], ctx: egui::Context) -> std::io::Result<Self> {
|
||||
let exe = std::env::current_exe()?;
|
||||
let mut child = Command::new(exe)
|
||||
.args(args)
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped())
|
||||
.spawn()?;
|
||||
|
||||
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::<ChildEvent>(&line) {
|
||||
if tx.send(ev).is_err() {
|
||||
break; // app gone
|
||||
}
|
||||
ctx.request_repaint();
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let stderr_tail = Arc::new(Mutex::new(Vec::<String>::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,
|
||||
rx,
|
||||
stderr_tail,
|
||||
})
|
||||
}
|
||||
|
||||
/// Whether the child is still running.
|
||||
pub fn is_alive(&mut self) -> bool {
|
||||
matches!(self.child.try_wait(), Ok(None))
|
||||
}
|
||||
|
||||
/// The last captured stderr lines, joined — for error display.
|
||||
pub fn stderr_tail(&self) -> String {
|
||||
self.stderr_tail.lock().unwrap().join("\n")
|
||||
}
|
||||
|
||||
/// Gracefully stop the child: SIGINT (so the host runs its ctrl-c teardown
|
||||
/// — tears down capture, closes the endpoint), with a ~2 s grace period
|
||||
/// before a hard kill. Idempotent.
|
||||
pub fn stop(&mut self) {
|
||||
if matches!(self.child.try_wait(), Ok(Some(_))) {
|
||||
return; // already exited
|
||||
}
|
||||
let _ = kill(Pid::from_raw(self.child.id() as i32), Signal::SIGINT);
|
||||
for _ in 0..40 {
|
||||
if matches!(self.child.try_wait(), Ok(Some(_))) {
|
||||
return;
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(50));
|
||||
}
|
||||
let _ = self.child.kill();
|
||||
let _ = self.child.wait();
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for ChildProc {
|
||||
fn drop(&mut self) {
|
||||
// Closing the window (dropping the app, hence the session) must not
|
||||
// orphan a live host child streaming to viewers.
|
||||
self.stop();
|
||||
}
|
||||
}
|
||||
|
||||
#[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::<ChildEvent>(&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_count_round_trips() {
|
||||
assert!(matches!(
|
||||
parse(Event::ViewerCount { active: 2, max: 4 }),
|
||||
ChildEvent::ViewerCount { active: 2, max: 4 }
|
||||
));
|
||||
}
|
||||
|
||||
#[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"
|
||||
));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user