feat(host): X11 capture backend + shared pipeline extraction
Extract the display-agnostic encode/mux tail out of wayland.rs into a new host/pipeline.rs: CaptureHandle + lifecycle, audio routing setup, the gst arg builder, the spawn, and Serve::bind now live there. Backends supply only their video-source element args plus a post-spawn hook (Wayland uses it to close its leaked pipewire fd; X11 passes a no-op). capture.rs collapses to a thin dispatcher; its CaptureHandle enum is gone. Add host/x11.rs: ximagesrc (use-damage=false show-pointer=true), whole root window by default or a single window via --window (xwininfo click-picker → xid). x11rb reads geometry for an info log, justifying the previously-vestigial dep. No portal, no fd dance — capture starts silently when the first viewer connects (the ticket is the access control). Viewer is display-agnostic and unchanged. Wire --no-hwencode for real (was a no-op): the shared tail now selects x264enc(tune=zerolatency,ultrafast)/I420 vs vah264enc/NV12 and switches the videoconvert target format to match. Applies to both backends. deps.rs: check_host_binaries now takes &HostOpts and checks shared elements for both backends, encoder by --no-hwencode, source per backend (pipewiresrc/ximagesrc), and xwininfo only when X11 + --window. Install hints added for x264enc, ximagesrc, xwininfo. Verified: warning-free build; smoke test still passes (tail unchanged); ximagesrc + both encoder tails produce mpv-decodable H.264 against an Xwayland root. Interactive cross-machine end-to-end pending. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.7
@@ -0,0 +1,255 @@
|
||||
//! Display-server-agnostic capture pipeline. The video *source* element is the
|
||||
//! only part that differs between Wayland (`pipewiresrc`, after a portal
|
||||
//! handshake) and X11 (`ximagesrc`); everything downstream — the videorate cap,
|
||||
//! the encoder, `h264parse`, `mpegtsmux`, the whole audio branch, the gst spawn,
|
||||
//! the [`Serve`] fanout binding, and the [`CaptureHandle`] lifecycle — is shared
|
||||
//! and lives here. Backends call [`spawn`] with just their source-element args.
|
||||
|
||||
use anyhow::{Context, Result, bail};
|
||||
use nix::sys::signal::{Signal, kill};
|
||||
use nix::unistd::Pid;
|
||||
use std::process::Stdio;
|
||||
use std::time::Duration;
|
||||
use tokio::process::{Child, Command};
|
||||
use tokio::time::timeout;
|
||||
|
||||
use super::audio::Routing;
|
||||
use super::serve::Serve;
|
||||
use crate::cli::HostOpts;
|
||||
|
||||
pub struct CaptureHandle {
|
||||
gst: Option<Child>,
|
||||
audio: Option<Routing>,
|
||||
serve: Option<Serve>,
|
||||
}
|
||||
|
||||
impl CaptureHandle {
|
||||
pub fn local_port(&self) -> u16 {
|
||||
self.serve
|
||||
.as_ref()
|
||||
.expect("serve is always Some until shutdown")
|
||||
.local_port()
|
||||
}
|
||||
|
||||
/// Graceful teardown: SIGTERM gst, give it ~1s to exit, then SIGKILL,
|
||||
/// unload audio routing (if any), then tear down the serve layer.
|
||||
/// The serve reader will see EOF on gst stdout and exit on its own;
|
||||
/// serve.shutdown() is the backstop.
|
||||
pub async fn shutdown(mut self) {
|
||||
if let Some(child) = self.gst.as_mut()
|
||||
&& let Some(pid) = child.id()
|
||||
{
|
||||
let _ = kill(Pid::from_raw(pid as i32), Signal::SIGTERM);
|
||||
}
|
||||
if let Some(child) = self.gst.as_mut() {
|
||||
let _ = timeout(Duration::from_millis(1000), child.wait()).await;
|
||||
let _ = child.start_kill();
|
||||
}
|
||||
if let Some(audio) = self.audio.take() {
|
||||
audio.shutdown();
|
||||
}
|
||||
if let Some(serve) = self.serve.take() {
|
||||
serve.shutdown().await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for CaptureHandle {
|
||||
fn drop(&mut self) {
|
||||
if let Some(child) = self.gst.as_mut() {
|
||||
let _ = child.start_kill();
|
||||
}
|
||||
// Routing's and Serve's own Drop impls handle the rest.
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawn the shared gst pipeline for a backend that supplies `source_args`
|
||||
/// (the video-source element + its properties, e.g. `["pipewiresrc", "fd=7",
|
||||
/// …]` or `["ximagesrc", "use-damage=false", …]`). `after_spawn` runs once,
|
||||
/// immediately after the gst child is launched — Wayland uses it to `close`
|
||||
/// the pipewire fd it leaked into the child; X11 passes a no-op.
|
||||
pub async fn spawn(
|
||||
opts: &HostOpts,
|
||||
source_args: Vec<String>,
|
||||
after_spawn: impl FnOnce(),
|
||||
) -> Result<CaptureHandle> {
|
||||
let (audio_routing, audio_device) = setup_audio(opts).await?;
|
||||
let args = build_args(&source_args, &audio_device, opts);
|
||||
|
||||
let mut gst_cmd = Command::new("gst-launch-1.0");
|
||||
gst_cmd
|
||||
.args(&args)
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::inherit());
|
||||
if std::env::var_os("PIXELPASS_GST_DEBUG").is_some() {
|
||||
gst_cmd.env("GST_DEBUG", "3");
|
||||
}
|
||||
let mut gst = gst_cmd.spawn().context("failed to spawn gst-launch-1.0")?;
|
||||
|
||||
// Backend-specific post-spawn cleanup (Wayland closes its leaked pw fd here,
|
||||
// once gst has inherited its own copy).
|
||||
after_spawn();
|
||||
|
||||
let gst_stdout = gst
|
||||
.stdout
|
||||
.take()
|
||||
.context("gst-launch-1.0 stdout pipe unavailable")?;
|
||||
|
||||
// Hand stdout to the serve layer, which binds the localhost HTTP listener
|
||||
// and runs the broadcast fanout. No demux/remux, no codec assumptions.
|
||||
let serve = Serve::bind(gst_stdout).await?;
|
||||
|
||||
Ok(CaptureHandle {
|
||||
gst: Some(gst),
|
||||
audio: audio_routing,
|
||||
serve: Some(serve),
|
||||
})
|
||||
}
|
||||
|
||||
/// Decide whether per-app audio routing is active and produce the `device=…`
|
||||
/// argument for `pulsesrc`. Routing activates when either `--app` is set
|
||||
/// (per-stream rerouting to a per-PID null-sink) or `PIXELPASS_AUDIO_VIA_NULL_SINK=1`
|
||||
/// is set (no app filter — captures everything via the null-sink, used for
|
||||
/// dogfooding the loopback path). Otherwise we capture the default sink's
|
||||
/// monitor (system audio out), not the default source (the mic).
|
||||
async fn setup_audio(opts: &HostOpts) -> Result<(Option<Routing>, String)> {
|
||||
let routing_requested =
|
||||
opts.app.is_some() || std::env::var_os("PIXELPASS_AUDIO_VIA_NULL_SINK").is_some();
|
||||
let audio_routing = if routing_requested {
|
||||
Some(
|
||||
Routing::start(opts)
|
||||
.await
|
||||
.context("audio routing setup failed")?,
|
||||
)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let audio_device = if let Some(r) = &audio_routing {
|
||||
format!("device={}.monitor", r.sink_name())
|
||||
} else {
|
||||
let default = default_audio_monitor().await?;
|
||||
format!("device={default}")
|
||||
};
|
||||
Ok((audio_routing, audio_device))
|
||||
}
|
||||
|
||||
/// Build the full gst-launch argument vector: MPEG-TS mux + fdsink, then the
|
||||
/// video branch (caller's `source` → videorate cap → encoder → h264parse →
|
||||
/// mux.), then the audio branch (pulsesrc → AAC → mux.). The encoder and the
|
||||
/// `videoconvert` target format are selected by `opts.no_hwencode`:
|
||||
/// hardware VAAPI wants NV12, software x264 wants I420.
|
||||
fn build_args(source: &[String], audio_device: &str, opts: &HostOpts) -> Vec<String> {
|
||||
let key_interval = (opts.framerate * 2).to_string();
|
||||
let bitrate = opts.bitrate.to_string();
|
||||
let framerate_caps = format!("video/x-raw,framerate={}/1", opts.framerate);
|
||||
|
||||
let (raw_format, encoder_args): (&str, Vec<String>) = if opts.no_hwencode {
|
||||
(
|
||||
"video/x-raw,format=I420",
|
||||
vec![
|
||||
"x264enc".into(),
|
||||
"tune=zerolatency".into(),
|
||||
"speed-preset=ultrafast".into(),
|
||||
format!("bitrate={bitrate}"),
|
||||
format!("key-int-max={key_interval}"),
|
||||
],
|
||||
)
|
||||
} else {
|
||||
(
|
||||
"video/x-raw,format=NV12",
|
||||
vec![
|
||||
"vah264enc".into(),
|
||||
"rate-control=cbr".into(),
|
||||
format!("bitrate={bitrate}"),
|
||||
format!("key-int-max={key_interval}"),
|
||||
],
|
||||
)
|
||||
};
|
||||
|
||||
// muxer + sink
|
||||
let mut args: Vec<String> = vec![
|
||||
"mpegtsmux".into(),
|
||||
"name=mux".into(),
|
||||
"!".into(),
|
||||
"queue".into(),
|
||||
"!".into(),
|
||||
"fdsink".into(),
|
||||
"fd=1".into(),
|
||||
];
|
||||
|
||||
// video branch — videorate caps to the target fps so we don't ship at the
|
||||
// monitor's refresh rate (e.g. 180Hz) and pile up frames in the demuxer
|
||||
// queue faster than realtime.
|
||||
args.extend(source.iter().cloned());
|
||||
args.extend([
|
||||
"!".into(),
|
||||
"videorate".into(),
|
||||
"!".into(),
|
||||
framerate_caps,
|
||||
"!".into(),
|
||||
"queue".into(),
|
||||
"!".into(),
|
||||
"videoconvert".into(),
|
||||
"!".into(),
|
||||
raw_format.into(),
|
||||
"!".into(),
|
||||
]);
|
||||
args.extend(encoder_args);
|
||||
args.extend([
|
||||
"!".into(),
|
||||
"h264parse".into(),
|
||||
"config-interval=-1".into(),
|
||||
"!".into(),
|
||||
"video/x-h264,stream-format=byte-stream,alignment=au".into(),
|
||||
"!".into(),
|
||||
"mux.".into(),
|
||||
]);
|
||||
|
||||
// audio branch — capture the default sink's MONITOR (system audio out),
|
||||
// not the default source (which is the mic).
|
||||
args.extend([
|
||||
"pulsesrc".into(),
|
||||
audio_device.to_string(),
|
||||
"do-timestamp=true".into(),
|
||||
"!".into(),
|
||||
"queue".into(),
|
||||
"!".into(),
|
||||
"audioconvert".into(),
|
||||
"!".into(),
|
||||
"audioresample".into(),
|
||||
"!".into(),
|
||||
"audio/x-raw,rate=48000,channels=2".into(),
|
||||
"!".into(),
|
||||
"avenc_aac".into(),
|
||||
"bitrate=128000".into(),
|
||||
"!".into(),
|
||||
"aacparse".into(),
|
||||
"!".into(),
|
||||
"mux.".into(),
|
||||
]);
|
||||
|
||||
args
|
||||
}
|
||||
|
||||
async fn default_audio_monitor() -> Result<String> {
|
||||
let output = Command::new("pactl")
|
||||
.arg("get-default-sink")
|
||||
.output()
|
||||
.await
|
||||
.context("failed to run `pactl get-default-sink` (install pulseaudio-utils or pipewire-pulse)")?;
|
||||
if !output.status.success() {
|
||||
bail!(
|
||||
"pactl get-default-sink failed: {}",
|
||||
String::from_utf8_lossy(&output.stderr).trim()
|
||||
);
|
||||
}
|
||||
let sink = String::from_utf8(output.stdout)
|
||||
.context("default sink name was not UTF-8")?
|
||||
.trim()
|
||||
.to_string();
|
||||
if sink.is_empty() {
|
||||
bail!("pactl get-default-sink returned no name (is a sound server running?)");
|
||||
}
|
||||
Ok(format!("{sink}.monitor"))
|
||||
}
|
||||
Reference in New Issue
Block a user