lesavka/server/src/video.rs

// server/src/video.rs

use anyhow::Context;
use gstreamer as gst;
use gstreamer_app as gst_app;
use gst::prelude::*;
use gst::log;
use lesavka_common::lesavka::VideoPacket;
use tokio_stream::wrappers::ReceiverStream;
use tonic::Status;
use tracing::{debug, enabled, Level};

static START: std::sync::OnceLock<gst::ClockTime> = std::sync::OnceLock::new();

pub async fn spawn_camera(
    dev: &str,
    id: u32,
    _max_bitrate_kbit: u32,
) -> anyhow::Result<ReceiverStream<Result<VideoPacket, Status>>> {
    gst::init().context("gst init")?;

    // IMPORTANT: keep one AU per buffer, include regular SPS/PPS
    let desc = format!(
        "v4l2src device={dev} io-mode=mmap ! \
         video/x-h264,stream-format=byte-stream,alignment=au ! \
         h264parse config-interval=1 ! \
         appsink name=sink emit-signals=true drop=true sync=false"
    );    

    let pipeline = gst::parse::launch(&desc)?
        .downcast::<gst::Pipeline>()
        .expect("pipeline down-cast");

    let sink = pipeline
        .by_name("sink")
        .expect("appsink")
        .dynamic_cast::<gst_app::AppSink>()
        .expect("appsink downcast");

    let (tx, rx) = tokio::sync::mpsc::channel(256);

    sink.set_callbacks(
        gst_app::AppSinkCallbacks::builder()
            // .new_sample(move |sink| {
            //     let sample = sink.pull_sample().map_err(|_| gst::FlowError::Eos)?;
            //     let buffer = sample.buffer().ok_or(gst::FlowError::Error)?;

            //     let map = buffer.map_readable().map_err(|_| gst::FlowError::Error)?;
            //     let pkt = VideoPacket {
            //         id,
            //         pts: buffer.pts().nseconds() / 1_000, // → µs
            //         data: map.as_slice().to_vec(),
            //     };
            //     // ignore back‑pressure: drop oldest if channel is full
            //     let _ = tx.try_send(Ok(pkt));
            //     Ok(gst::FlowSuccess::Ok)
            // })
            // .build(),
            .new_sample(move |sink| {
                let sample = sink.pull_sample().map_err(|_| gst::FlowError::Eos)?;
                let buffer = sample.buffer().ok_or(gst::FlowError::Error)?;
                let origin = *START.get_or_init(|| buffer.pts().unwrap_or(gst::ClockTime::ZERO));

                let pts_us = buffer
                    .pts()
                    .unwrap_or(gst::ClockTime::ZERO)
                    .saturating_sub(origin)
                    .nseconds() / 1_000;

                let map = buffer.map_readable().map_err(|_| gst::FlowError::Error)?;

                if enabled!(Level::DEBUG) {
                    if let Some(slice) = map.as_slice().get(0..5) {
                        match slice[4] & 0b1_1111 {
                            // 0x07 = SPS, 0x05 = IDR (Annex-B “byte-stream”)
                            0x07 | 0x05 => {
                                debug!(
                                    "🎞️  monitor {id}: got NAL {:02X}, {} bytes",
                                    slice[4],
                                    map.as_slice().len()
                                );
                            }
                            _ => {}
                        }
                    }
                }

                let pkt = VideoPacket {
                    id,
                    pts: pts_us,
                    data: map.as_slice().to_vec(),
                };
                let _ = tx.try_send(Ok(pkt));        // drop on overflow
                Ok(gst::FlowSuccess::Ok)
            })
            .build(),
    );

    // gst::debug_remove_default_log_function();
    // gst::debug_add_default_log_function(|lvl, cat, msg| {
    //     println!("[GST] {lvl:?} {cat}: {msg}");
    // });
    // std::env::set_var("GST_DEBUG", "v4l2src:4,h264parse:3");

    pipeline.set_state(gst::State::Playing)?;

    Ok(ReceiverStream::new(rx))
}