feat(capture): add the PipeWire capture of the gamescope node

Finds the gamescope node, negotiates a small BGRx MemFd format through
gamescope's requested_size extension, converts each buffer and publishes it
into the frame slot. Runs on the PipeWire main loop in the main thread.

src/capture.rs

@@ -0,0 +1,758 @@
+//! PipeWire side (spec 4.2): finds the gamescope node, negotiates a small BGRx MemFd format
+//! through gamescope's `requested_size` extension, converts buffers and publishes frames.
+//!
+//! Everything here runs on the PipeWire main loop in the main thread.
+
+use std::cell::{Cell, RefCell};
+use std::collections::{BTreeMap, VecDeque};
+use std::fmt::{self, Write as _};
+use std::io::Cursor;
+use std::rc::Rc;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use anyhow::{anyhow, Context as _};
+use pipewire as pw;
+use pw::properties::properties;
+use pw::spa;
+use pw::stream::{Stream, StreamFlags, StreamListener, StreamRc, StreamState};
+use spa::buffer::meta::{MetaHeader, MetaHeaderFlags};
+use spa::buffer::ChunkFlags;
+use spa::param::format::{FormatProperties, MediaSubtype, MediaType};
+use spa::param::video::{VideoFormat, VideoInfoRaw};
+use spa::param::ParamType;
+use spa::pod::deserialize::PodDeserializer;
+use spa::pod::serialize::PodSerializer;
+use spa::pod::{ChoiceValue, Object, Pod, Property, PropertyFlags, Value};
+use spa::utils::{Choice, ChoiceEnum, ChoiceFlags, Direction, Fraction, Id, Rectangle, SpaTypes};
+use tracing::{debug, error, info, warn};
+
+use crate::config::Size;
+use crate::convert::{fit_within, BgrxFrame, Converter};
+use crate::pacing::RateLimit;
+use crate::slot::{Frame, FrameSlot};
+use crate::stats::{self, Stats};
+
+/// gamescope's `SPA_FORMAT_VIDEO_requested_size` (`src/pipewire_gamescope.hpp`).
+pub const SPA_FORMAT_VIDEO_REQUESTED_SIZE: u32 = 0x70000;
+/// gamescope's `SPA_FORMAT_VIDEO_gamescope_focus_appid`. Deliberately never set.
+pub const SPA_FORMAT_VIDEO_GAMESCOPE_FOCUS_APPID: u32 = 0x70001;
+/// gamescope's `SPA_META_requested_size_scale`.
+pub const SPA_META_REQUESTED_SIZE_SCALE: u32 = 0x70000;
+
+const STREAM_NODE_NAME: &str = "gamescope-hyperhdr";
+const SOURCE_MEDIA_CLASS: &str = "Video/Source";
+const RETRY_DELAY: Duration = Duration::from_secs(1);
+/// Period of the housekeeping timer that applies deferred teardowns and (re)creates the stream.
+/// A stream must not be destroyed from its own callbacks, and `pw::channel` cannot defer the
+/// work: its receiver callback holds the channel lock, so sending from there deadlocks.
+const TICK: Duration = Duration::from_millis(100);
+/// Explain the silence when a streaming stream has not delivered a frame for this long.
+const NO_FRAMES_HINT_AFTER: Duration = Duration::from_secs(5);
+
+#[derive(Debug, Clone)]
+pub struct CaptureConfig {
+    /// `node.name` of the source.
+    pub target: String,
+    /// Size requested from gamescope (`size × supersample`).
+    pub requested: Size,
+    /// Frames are downscaled to fit into this size.
+    pub output_max: Size,
+}
+
+struct Teardown {
+    generation: u64,
+    reason: String,
+    retry_delay: Option<Duration>,
+}
+
+type Teardowns = Rc<RefCell<VecDeque<Teardown>>>;
+
+/// Progress towards the first frame of a stream, shared by its callbacks and the housekeeping
+/// timer. gamescope renders a frame for PipeWire only when the picture changes, so on a static
+/// screen the first frame can take arbitrarily long.
+#[derive(Debug, Default)]
+struct FirstFrame {
+    /// When the stream first reached Streaming.
+    streaming_since: Cell<Option<Instant>>,
+    received: Cell<bool>,
+    hinted: Cell<bool>,
+}
+
+struct Active {
+    node_id: u32,
+    serial: u64,
+    generation: u64,
+    first_frame: Rc<FirstFrame>,
+    // Dropped before the stream, so no callbacks run during its destruction.
+    _listener: StreamListener<StreamData>,
+    _stream: StreamRc,
+}
+
+struct State {
+    cfg: CaptureConfig,
+    core: pw::core::CoreRc,
+    slot: Arc<FrameSlot>,
+    stats: Arc<Stats>,
+    teardowns: Teardowns,
+    /// Matching source nodes: global id → object.serial.
+    candidates: BTreeMap<u32, u64>,
+    active: Option<Active>,
+    retry_at: Option<Instant>,
+    generation: u64,
+}
+
+/// Runs the PipeWire main loop until SIGINT/SIGTERM (`Ok`) or a fatal PipeWire error (`Err`).
+///
+/// `on_ready` runs once the signal handlers are installed; start other threads there.
+/// PipeWire receives signals through a signalfd and blocks them only in the calling thread,
+/// so a thread spawned earlier would still get the default (terminating) disposition.
+pub fn run(
+    cfg: CaptureConfig,
+    slot: Arc<FrameSlot>,
+    stats: Arc<Stats>,
+    on_ready: impl FnOnce() -> anyhow::Result<()>,
+) -> anyhow::Result<()> {
+    pw::init();
+    let mainloop =
+        pw::main_loop::MainLoopRc::new(None).context("cannot create the PipeWire main loop")?;
+    let _signals = [
+        (pw::loop_::Signal::INT, "SIGINT"),
+        (pw::loop_::Signal::TERM, "SIGTERM"),
+    ]
+    .map(|(signal, name)| {
+        let mainloop_weak = mainloop.downgrade();
+        mainloop.loop_().add_signal_local(signal, move || {
+            info!("{name} received, shutting down");
+            if let Some(mainloop) = mainloop_weak.upgrade() {
+                mainloop.quit();
+            }
+        })
+    });
+    on_ready()?;
+
+    let context =
+        pw::context::ContextRc::new(&mainloop, None).context("cannot create a PipeWire context")?;
+    let core = context
+        .connect_rc(None)
+        .context("cannot connect to the PipeWire daemon")?;
+    let registry = core
+        .get_registry_rc()
+        .context("cannot get the PipeWire registry")?;
+
+    let fatal: Rc<RefCell<Option<String>>> = Rc::default();
+    let _core_listener = {
+        let mainloop_weak = mainloop.downgrade();
+        let fatal = fatal.clone();
+        core.add_listener_local()
+            .error(move |id, seq, res, message| {
+                let broken_pipe = std::io::Error::from_raw_os_error(-res).kind()
+                    == std::io::ErrorKind::BrokenPipe;
+                if id == pw::sys::PW_ID_CORE && broken_pipe {
+                    error!(seq, res, "lost connection to PipeWire: {message}");
+                    *fatal.borrow_mut() =
+                        Some(format!("lost connection to PipeWire: {message} ({res})"));
+                    if let Some(mainloop) = mainloop_weak.upgrade() {
+                        mainloop.quit();
+                    }
+                } else if id == pw::sys::PW_ID_CORE {
+                    // E.g. "unknown resource" after the server destroyed our stream's node
+                    // because its target disappeared.
+                    warn!(seq, res, "PipeWire core error: {message}");
+                } else {
+                    // Stream errors are also reported through its state_changed event.
+                    debug!(id, seq, res, "PipeWire error: {message}");
+                }
+            })
+            .register()
+    };
+
+    let state = Rc::new(RefCell::new(State {
+        cfg,
+        core: core.clone(),
+        slot,
+        stats,
+        teardowns: Teardowns::default(),
+        candidates: BTreeMap::new(),
+        active: None,
+        retry_at: None,
+        generation: 0,
+    }));
+
+    let _registry_listener = {
+        let (on_add, on_remove) = (state.clone(), state.clone());
+        registry
+            .add_listener_local()
+            .global(move |global| {
+                if global.type_ != pw::types::ObjectType::Node {
+                    return;
+                }
+                let Some(props) = global.props else {
+                    return;
+                };
+                let mut s = on_add.borrow_mut();
+                if props.get("node.name") != Some(s.cfg.target.as_str()) {
+                    return;
+                }
+                let class = props.get("media.class");
+                if class != Some(SOURCE_MEDIA_CLASS) {
+                    debug!(
+                        id = global.id,
+                        "ignoring node '{}' with media.class {class:?}", s.cfg.target
+                    );
+                    return;
+                }
+                let serial = props
+                    .get("object.serial")
+                    .and_then(|v| v.parse().ok())
+                    .unwrap_or(u64::from(global.id));
+                info!(
+                    id = global.id,
+                    serial, "found source node '{}'", s.cfg.target
+                );
+                // The next tick picks the newest candidate, after the registry burst.
+                s.candidates.insert(global.id, serial);
+            })
+            .global_remove(move |id| {
+                let mut s = on_remove.borrow_mut();
+                if let Some(serial) = s.candidates.remove(&id) {
+                    info!(id, serial, "source node removed");
+                }
+                if let Some(active) = s.active.as_ref().filter(|a| a.node_id == id) {
+                    let teardown = Teardown {
+                        generation: active.generation,
+                        reason: "source node removed".to_owned(),
+                        retry_delay: None,
+                    };
+                    s.teardowns.borrow_mut().push_back(teardown);
+                }
+            })
+            .register()
+    };
+
+    let tick = {
+        let state = state.clone();
+        mainloop.loop_().add_timer(move |_| housekeeping(&state))
+    };
+    let _ = tick.update_timer(Some(TICK), Some(TICK));
+
+    info!(
+        "waiting for a '{}' {SOURCE_MEDIA_CLASS} node",
+        state.borrow().cfg.target
+    );
+    mainloop.run();
+
+    let active = state.borrow_mut().active.take();
+    drop(active);
+
+    let fatal = fatal.borrow_mut().take();
+    match fatal {
+        Some(reason) => Err(anyhow!(reason)),
+        None => Ok(()),
+    }
+}
+
+fn housekeeping(state: &Rc<RefCell<State>>) {
+    let teardowns = state.borrow().teardowns.clone();
+    loop {
+        let next = teardowns.borrow_mut().pop_front();
+        match next {
+            Some(t) => teardown(state, t),
+            None => break,
+        }
+    }
+    reconcile(state);
+    hint_missing_frames(state);
+}
+
+fn hint_missing_frames(state: &Rc<RefCell<State>>) {
+    let s = state.borrow();
+    let Some(active) = &s.active else {
+        return;
+    };
+    let first_frame = &active.first_frame;
+    if first_frame.received.get() || first_frame.hinted.get() {
+        return;
+    }
+    if first_frame
+        .streaming_since
+        .get()
+        .is_some_and(|since| since.elapsed() >= NO_FRAMES_HINT_AFTER)
+    {
+        first_frame.hinted.set(true);
+        info!(
+            node_id = active.node_id,
+            "stream is streaming, but no frames for {} s: gamescope sends frames only when the \
+             picture changes; the HyperHDR connection opens with the first frame",
+            NO_FRAMES_HINT_AFTER.as_secs()
+        );
+    }
+}
+
+fn reconcile(state: &Rc<RefCell<State>>) {
+    let mut s = state.borrow_mut();
+    if s.active.is_some()
+        || s.candidates.is_empty()
+        || s.retry_at.is_some_and(|at| Instant::now() < at)
+    {
+        return;
+    }
+    let Some((&node_id, &serial)) = s.candidates.iter().max_by_key(|(_, serial)| **serial) else {
+        return;
+    };
+    if s.candidates.len() > 1 {
+        warn!(
+            count = s.candidates.len(),
+            node_id,
+            serial,
+            "several '{}' {SOURCE_MEDIA_CLASS} nodes exist; using the newest (highest object.serial)",
+            s.cfg.target
+        );
+    }
+    s.generation += 1;
+    let generation = s.generation;
+    match connect_stream(&s, node_id, serial, generation) {
+        Ok(active) => {
+            info!(node_id, serial, requested_size = %s.cfg.requested, "stream created, linking to the source");
+            s.active = Some(active);
+            s.retry_at = None;
+        }
+        Err(e) => {
+            error!(node_id, serial, "cannot create the stream: {e:#}");
+            s.retry_at = Some(Instant::now() + RETRY_DELAY);
+        }
+    }
+}
+
+fn teardown(state: &Rc<RefCell<State>>, t: Teardown) {
+    let active = {
+        let mut s = state.borrow_mut();
+        if s.active.as_ref().map(|a| a.generation) != Some(t.generation) {
+            return;
+        }
+        s.retry_at = t.retry_delay.map(|d| Instant::now() + d);
+        s.active.take()
+    };
+    if let Some(active) = active {
+        info!(
+            node_id = active.node_id,
+            serial = active.serial,
+            "destroying stream: {}",
+            t.reason
+        );
+        drop(active);
+    }
+    state.borrow().slot.source_gone();
+}
+
+fn connect_stream(s: &State, node_id: u32, serial: u64, generation: u64) -> anyhow::Result<Active> {
+    let props = properties! {
+        *pw::keys::MEDIA_TYPE => "Video",
+        *pw::keys::MEDIA_CATEGORY => "Capture",
+        *pw::keys::MEDIA_ROLE => "Screen",
+        *pw::keys::NODE_NAME => STREAM_NODE_NAME,
+        *pw::keys::NODE_DESCRIPTION => "gamescope → HyperHDR",
+        "target.object" => serial.to_string(),
+        "node.dont-fallback" => "true",
+        *pw::keys::NODE_DONT_RECONNECT => "true",
+        "node.dont-move" => "true",
+    };
+    let stream =
+        StreamRc::new(s.core.clone(), STREAM_NODE_NAME, props).context("pw_stream_new failed")?;
+
+    let first_frame = Rc::new(FirstFrame::default());
+    let data = StreamData {
+        node_id,
+        generation,
+        requested: s.cfg.requested,
+        output_max: s.cfg.output_max,
+        slot: s.slot.clone(),
+        stats: s.stats.clone(),
+        teardowns: s.teardowns.clone(),
+        first_frame: first_frame.clone(),
+        format: None,
+        format_events: 0,
+        first_frame_logged: false,
+        warned_ignored: false,
+        converter: Converter::new(),
+        frame: Frame::default(),
+        bounds_log: RateLimit::new(Duration::from_secs(10)),
+    };
+    let listener = stream
+        .add_local_listener_with_user_data(data)
+        .state_changed(|_, d, old, new| on_state_changed(d, old, new))
+        .param_changed(|_, d, id, param| on_param_changed(d, id, param))
+        .process(on_process)
+        .register()
+        .context("cannot register stream listener")?;
+
+    let pod = enum_format_pod(s.cfg.requested);
+    if tracing::enabled!(tracing::Level::DEBUG) {
+        debug!("EnumFormat pod ({} bytes): {}", pod.len(), Hex(&pod));
+    }
+    let mut params = [Pod::from_bytes(&pod).ok_or_else(|| anyhow!("invalid EnumFormat pod"))?];
+    stream
+        .connect(
+            Direction::Input,
+            None,
+            StreamFlags::AUTOCONNECT | StreamFlags::MAP_BUFFERS,
+            &mut params,
+        )
+        .context("pw_stream_connect failed")?;
+
+    Ok(Active {
+        node_id,
+        serial,
+        generation,
+        first_frame,
+        _listener: listener,
+        _stream: stream,
+    })
+}
+
+/// The single EnumFormat object offered to gamescope (spec 4.2.4).
+pub fn enum_format_pod(requested: Size) -> Vec<u8> {
+    let rect = |width, height| Rectangle { width, height };
+    let requested_rect = rect(requested.width, requested.height);
+    let object = Object {
+        type_: SpaTypes::ObjectParamFormat.as_raw(),
+        id: ParamType::EnumFormat.as_raw(),
+        properties: vec![
+            Property::new(
+                FormatProperties::MediaType.as_raw(),
+                Value::Id(Id(MediaType::Video.as_raw())),
+            ),
+            Property::new(
+                FormatProperties::MediaSubtype.as_raw(),
+                Value::Id(Id(MediaSubtype::Raw.as_raw())),
+            ),
+            // BGRx only, and no modifier: that rules out gamescope's DMA-BUF offers
+            // (their modifier is MANDATORY) and NV12.
+            Property::new(
+                FormatProperties::VideoFormat.as_raw(),
+                Value::Id(Id(VideoFormat::BGRx.as_raw())),
+            ),
+            Property::new(
+                FormatProperties::VideoSize.as_raw(),
+                Value::Choice(ChoiceValue::Rectangle(Choice(
+                    ChoiceFlags::empty(),
+                    ChoiceEnum::Range {
+                        default: requested_rect,
+                        min: rect(1, 1),
+                        max: rect(16384, 16384),
+                    },
+                ))),
+            ),
+            // Must include 0/1: gamescope offers a variable frame rate.
+            Property::new(
+                FormatProperties::VideoFramerate.as_raw(),
+                Value::Choice(ChoiceValue::Fraction(Choice(
+                    ChoiceFlags::empty(),
+                    ChoiceEnum::Range {
+                        default: Fraction { num: 0, denom: 1 },
+                        min: Fraction { num: 0, denom: 1 },
+                        max: Fraction {
+                            num: 1000,
+                            denom: 1,
+                        },
+                    },
+                ))),
+            ),
+            // A fixed value, not a choice: gamescope renders into a frame that fits it.
+            Property {
+                key: SPA_FORMAT_VIDEO_REQUESTED_SIZE,
+                flags: PropertyFlags::empty(),
+                value: Value::Rectangle(requested_rect),
+            },
+        ],
+    };
+    PodSerializer::serialize(Cursor::new(Vec::new()), &Value::Object(object))
+        .expect("serializing into a Vec cannot fail")
+        .0
+        .into_inner()
+}
+
+/// Reads gamescope's `requested_size` from a Format pod.
+pub fn requested_size_of(format: &Pod) -> Option<Rectangle> {
+    let (_, Value::Object(object)) =
+        PodDeserializer::deserialize_any_from(format.as_bytes()).ok()?
+    else {
+        return None;
+    };
+    let prop = object
+        .properties
+        .into_iter()
+        .find(|p| p.key == SPA_FORMAT_VIDEO_REQUESTED_SIZE)?;
+    match prop.value {
+        Value::Rectangle(r) => Some(r),
+        Value::Choice(ChoiceValue::Rectangle(Choice(
+            _,
+            ChoiceEnum::None(r)
+            | ChoiceEnum::Range { default: r, .. }
+            | ChoiceEnum::Step { default: r, .. },
+        ))) => Some(r),
+        _ => None,
+    }
+}
+
+struct StreamData {
+    node_id: u32,
+    generation: u64,
+    requested: Size,
+    output_max: Size,
+    slot: Arc<FrameSlot>,
+    stats: Arc<Stats>,
+    teardowns: Teardowns,
+    first_frame: Rc<FirstFrame>,
+    /// Negotiated BGRx size, `None` until a usable format is negotiated.
+    format: Option<Size>,
+    format_events: u32,
+    first_frame_logged: bool,
+    warned_ignored: bool,
+    converter: Converter,
+    frame: Frame,
+    bounds_log: RateLimit,
+}
+
+impl StreamData {
+    fn bounds_error(&mut self, what: impl fmt::Display) {
+        stats::inc(&self.stats.bounds_errors);
+        if let Some(suppressed) = self.bounds_log.check(Instant::now()) {
+            warn!(suppressed, "skipping invalid buffer: {what}");
+        }
+    }
+}
+
+fn on_state_changed(d: &mut StreamData, old: StreamState, new: StreamState) {
+    let reason = match new {
+        StreamState::Error(e) => {
+            warn!(node_id = d.node_id, "stream error: {e}");
+            format!("stream error: {e}")
+        }
+        StreamState::Unconnected => {
+            info!(node_id = d.node_id, "stream state {old:?} -> Unconnected");
+            "stream unconnected".to_owned()
+        }
+        new => {
+            info!(node_id = d.node_id, "stream state {old:?} -> {new:?}");
+            // Renegotiation toggles Streaming/Paused; count from the first Streaming.
+            if new == StreamState::Streaming && d.first_frame.streaming_since.get().is_none() {
+                d.first_frame.streaming_since.set(Some(Instant::now()));
+            }
+            return;
+        }
+    };
+    d.teardowns.borrow_mut().push_back(Teardown {
+        generation: d.generation,
+        reason,
+        retry_delay: Some(RETRY_DELAY),
+    });
+}
+
+fn on_param_changed(d: &mut StreamData, id: u32, param: Option<&Pod>) {
+    if id != ParamType::Format.as_raw() {
+        return;
+    }
+    let Some(param) = param else {
+        info!("format cleared");
+        d.format = None;
+        return;
+    };
+    d.format_events += 1;
+    d.first_frame_logged = false;
+
+    let mut info = VideoInfoRaw::new();
+    if let Err(e) = info.parse(param) {
+        error!("cannot parse the negotiated format: {e:?}");
+        d.format = None;
+        return;
+    }
+    if tracing::enabled!(tracing::Level::DEBUG) {
+        if let Ok((_, value)) = PodDeserializer::deserialize_any_from(param.as_bytes()) {
+            debug!("Format pod: {value:?}");
+        }
+    }
+    if info.format() != VideoFormat::BGRx {
+        error!(
+            "negotiated unsupported video format {:?}, ignoring buffers",
+            info.format()
+        );
+        d.format = None;
+        return;
+    }
+    let size = info.size();
+    let framerate = info.framerate();
+    let requested = requested_size_of(param).map_or_else(
+        || "absent".to_owned(),
+        |r| format!("{}x{}", r.width, r.height),
+    );
+    info!(
+        event = d.format_events,
+        "format: BGRx {}x{}, expected stride {}, framerate {}/{}, requested_size {requested}",
+        size.width,
+        size.height,
+        size.width.saturating_mul(4).next_multiple_of(4),
+        framerate.num,
+        framerate.denom,
+    );
+    d.format = Some(Size::new(size.width, size.height));
+}
+
+fn on_process(stream: &Stream, d: &mut StreamData) {
+    // Drain the queue: only the newest buffer is used, older ones are requeued on drop.
+    let mut newest = None;
+    while let Some(buffer) = stream.dequeue_buffer() {
+        newest = Some(buffer);
+    }
+    let Some(mut buffer) = newest else {
+        return;
+    };
+    let Some(size) = d.format else {
+        return;
+    };
+
+    let header_corrupted = buffer
+        .find_meta::<MetaHeader>()
+        .is_some_and(|h| h.flags().contains(MetaHeaderFlags::CORRUPTED));
+    let Some(data) = buffer.datas_mut().first_mut() else {
+        d.bounds_error("buffer has no data blocks");
+        return;
+    };
+    let chunk = data.chunk();
+    let (offset, chunk_size, stride) = (chunk.offset(), chunk.size(), chunk.stride());
+    // Buffers without data show up around renegotiation; they are skipped like corrupted ones.
+    if header_corrupted || chunk.flags().contains(ChunkFlags::CORRUPTED) || chunk_size == 0 {
+        stats::inc(&d.stats.corrupted);
+        return;
+    }
+    let data_type = data.type_();
+    let Some(mapped) = data.data() else {
+        d.bounds_error(format_args!(
+            "buffer is not mapped (data type {data_type:?})"
+        ));
+        return;
+    };
+    let mapped_len = mapped.len();
+
+    let started = Instant::now();
+    let bgrx = match BgrxFrame::new(
+        mapped,
+        offset as usize,
+        chunk_size as usize,
+        usize::try_from(stride).unwrap_or(0),
+        size.width as usize,
+        size.height as usize,
+    ) {
+        Ok(frame) => frame,
+        Err(e) => {
+            d.bounds_error(format_args!(
+                "{e} (chunk offset {offset}, size {chunk_size}, stride {stride}; {mapped_len} bytes mapped)"
+            ));
+            return;
+        }
+    };
+
+    let requested = d.requested;
+    if !d.warned_ignored
+        && (size.width > requested.width.saturating_mul(2)
+            || size.height > requested.height.saturating_mul(2))
+    {
+        d.warned_ignored = true;
+        warn!(
+            "gamescope seems to ignore requested_size {requested}: receiving {size} frames, downscaling on the CPU"
+        );
+    }
+
+    let (width, height) = fit_within(
+        size.width,
+        size.height,
+        d.output_max.width,
+        d.output_max.height,
+    );
+    d.converter
+        .convert(&bgrx, width as usize, height as usize, &mut d.frame.rgb);
+    d.frame.width = width;
+    d.frame.height = height;
+    d.stats.record_convert(started.elapsed());
+    stats::inc(&d.stats.frames_received);
+
+    if !d.first_frame.received.replace(true) && d.first_frame.hinted.get() {
+        if let Some(since) = d.first_frame.streaming_since.get() {
+            info!(
+                "first frame arrived {:.1?} after the stream started streaming",
+                since.elapsed()
+            );
+        }
+    }
+    if !d.first_frame_logged {
+        d.first_frame_logged = true;
+        info!(
+            "receiving frames: {size} stride {stride} offset {offset} chunk {chunk_size} bytes, sending {width}x{height}"
+        );
+    }
+    if d.slot.publish(&mut d.frame) {
+        stats::inc(&d.stats.superseded);
+    }
+}
+
+struct Hex<'a>(&'a [u8]);
+
+impl fmt::Display for Hex<'_> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        for (i, b) in self.0.iter().enumerate() {
+            if i > 0 && i % 4 == 0 {
+                f.write_char(' ')?;
+            }
+            write!(f, "{b:02x}")?;
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn enum_format_pod_contents() {
+        let bytes = enum_format_pod(Size::new(640, 360));
+        let pod = Pod::from_bytes(&bytes).unwrap();
+        let (_, value) = PodDeserializer::deserialize_any_from(pod.as_bytes()).unwrap();
+        let Value::Object(object) = value else {
+            panic!("not an object");
+        };
+        assert_eq!(object.type_, SpaTypes::ObjectParamFormat.as_raw());
+        assert_eq!(object.id, ParamType::EnumFormat.as_raw());
+
+        let keys: Vec<u32> = object.properties.iter().map(|p| p.key).collect();
+        assert!(!keys.contains(&FormatProperties::VideoModifier.as_raw()));
+        assert!(!keys.contains(&SPA_FORMAT_VIDEO_GAMESCOPE_FOCUS_APPID));
+
+        let prop = |key: u32| object.properties.iter().find(|p| p.key == key).unwrap();
+        assert_eq!(
+            prop(FormatProperties::VideoFormat.as_raw()).value,
+            Value::Id(Id(VideoFormat::BGRx.as_raw()))
+        );
+        let custom = prop(SPA_FORMAT_VIDEO_REQUESTED_SIZE);
+        assert_eq!(custom.flags, PropertyFlags::empty());
+        assert_eq!(
+            custom.value,
+            Value::Rectangle(Rectangle {
+                width: 640,
+                height: 360
+            })
+        );
+        assert!(matches!(
+            prop(FormatProperties::VideoSize.as_raw()).value,
+            Value::Choice(ChoiceValue::Rectangle(Choice(_, ChoiceEnum::Range { .. })))
+        ));
+
+        assert_eq!(
+            requested_size_of(pod),
+            Some(Rectangle {
+                width: 640,
+                height: 360
+            })
+        );
+    }
+}

src/lib.rs

@@ -1,6 +1,7 @@
 //! Captures gamescope's PipeWire stream at a small `requested_size` and forwards it to
 //! HyperHDR's flatbuffers server.
 
+pub mod capture;
 pub mod config;
 pub mod convert;
 pub mod hyperhdr;