lesavka/client/src/launcher/preview.rs

#[cfg(not(coverage))]
use anyhow::{Context, Result};
#[cfg(not(coverage))]
use gstreamer as gst;
#[cfg(not(coverage))]
use gstreamer::prelude::{Cast, ElementExt, GstBinExt};
#[cfg(not(coverage))]
use gstreamer_app as gst_app;
#[cfg(not(coverage))]
use gtk::{gdk, glib};
#[cfg(not(coverage))]
use lesavka_common::lesavka::{relay_client::RelayClient, MonitorRequest, VideoPacket};
#[cfg(not(coverage))]
use std::sync::atomic::{AtomicBool, Ordering};
#[cfg(not(coverage))]
use std::sync::{Arc, Mutex};
#[cfg(not(coverage))]
use std::time::Duration;
#[cfg(not(coverage))]
use tonic::{transport::Channel, Request};
#[cfg(not(coverage))]
use tracing::{debug, warn};

#[cfg(not(coverage))]
const PREVIEW_WIDTH: i32 = 640;
#[cfg(not(coverage))]
const PREVIEW_HEIGHT: i32 = 360;
#[cfg(not(coverage))]
const PREVIEW_IDLE_STATUS: &str = "Connect relay to preview.";

#[cfg(not(coverage))]
pub struct LauncherPreview {
    server_addr: Arc<Mutex<String>>,
    feeds: [PreviewFeed; 2],
}

#[cfg(not(coverage))]
#[derive(Clone)]
pub struct PreviewBinding {
    enabled: Arc<AtomicBool>,
    alive: Arc<AtomicBool>,
}

#[cfg(not(coverage))]
impl LauncherPreview {
    pub fn new(server_addr: String) -> Result<Self> {
        gst::init().context("initialising preview gstreamer")?;
        let server_addr = Arc::new(Mutex::new(server_addr));
        Ok(Self {
            server_addr: Arc::clone(&server_addr),
            feeds: [
                PreviewFeed::spawn(Arc::clone(&server_addr), 0)?,
                PreviewFeed::spawn(server_addr, 1)?,
            ],
        })
    }

    pub fn set_server_addr(&self, server_addr: String) {
        if let Ok(mut slot) = self.server_addr.lock() {
            *slot = server_addr;
        }
    }

    pub fn set_session_active(&self, active: bool) {
        for feed in &self.feeds {
            feed.set_active(active);
        }
    }

    pub fn install_on_picture(
        &self,
        monitor_id: usize,
        picture: &gtk::Picture,
        status_label: &gtk::Label,
    ) -> Option<PreviewBinding> {
        self.feeds
            .get(monitor_id)
            .map(|feed| feed.install_on_picture(picture, status_label))
    }
}

#[cfg(not(coverage))]
impl PreviewBinding {
    pub fn set_enabled(&self, enabled: bool) {
        self.enabled.store(enabled, Ordering::Relaxed);
    }

    pub fn close(&self) {
        self.alive.store(false, Ordering::Relaxed);
    }
}

#[cfg(not(coverage))]
struct PreviewFeed {
    shared: Arc<Mutex<SharedPreviewState>>,
    active: Arc<AtomicBool>,
}

#[cfg(not(coverage))]
struct SharedPreviewState {
    latest: Option<PreviewFrame>,
    status: String,
    generation: u64,
    clear_picture: bool,
}

#[cfg(not(coverage))]
impl SharedPreviewState {
    fn new() -> Self {
        Self {
            latest: None,
            status: PREVIEW_IDLE_STATUS.to_string(),
            generation: 1,
            clear_picture: true,
        }
    }

    fn set_status(&mut self, status: impl Into<String>, clear_picture: bool) {
        let status = status.into();
        let changed = self.status != status || clear_picture;
        self.status = status;
        if clear_picture {
            self.latest = None;
            self.clear_picture = true;
        }
        if changed {
            self.generation = self.generation.saturating_add(1);
        }
    }

    fn push_frame(&mut self, frame: PreviewFrame) {
        self.latest = Some(frame);
        self.clear_picture = false;
        if self.status != "Live" {
            self.status = "Live".to_string();
            self.generation = self.generation.saturating_add(1);
        }
    }
}

#[cfg(not(coverage))]
impl PreviewFeed {
    fn spawn(server_addr: Arc<Mutex<String>>, monitor_id: u32) -> Result<Self> {
        let shared = Arc::new(Mutex::new(SharedPreviewState::new()));
        let active = Arc::new(AtomicBool::new(false));
        let shared_state = Arc::clone(&shared);
        let active_flag = Arc::clone(&active);
        std::thread::spawn(move || {
            if let Err(err) = run_preview_feed(server_addr, monitor_id, active_flag, shared_state) {
                warn!(monitor_id, ?err, "launcher preview feed exited");
            }
        });
        Ok(Self { shared, active })
    }

    fn set_active(&self, active: bool) {
        self.active.store(active, Ordering::Relaxed);
        if !active {
            self.replace_status(PREVIEW_IDLE_STATUS, true);
        }
    }

    fn replace_status(&self, status: impl Into<String>, clear_picture: bool) {
        if let Ok(mut shared) = self.shared.lock() {
            shared.set_status(status, clear_picture);
        }
    }

    fn install_on_picture(
        &self,
        picture: &gtk::Picture,
        status_label: &gtk::Label,
    ) -> PreviewBinding {
        let picture = picture.clone();
        let status_label = status_label.clone();
        let shared = Arc::clone(&self.shared);
        let enabled = Arc::new(AtomicBool::new(true));
        let alive = Arc::new(AtomicBool::new(true));
        let enabled_flag = Arc::clone(&enabled);
        let alive_flag = Arc::clone(&alive);
        let mut last_generation = 0_u64;
        glib::timeout_add_local(Duration::from_millis(120), move || {
            if !alive_flag.load(Ordering::Relaxed) {
                return glib::ControlFlow::Break;
            }
            if !enabled_flag.load(Ordering::Relaxed) {
                return glib::ControlFlow::Continue;
            }

            let (frame, status, generation, clear_picture) = match shared.lock() {
                Ok(mut slot) => {
                    let frame = slot.latest.take();
                    let status = slot.status.clone();
                    let generation = slot.generation;
                    let clear_picture = slot.clear_picture;
                    slot.clear_picture = false;
                    (frame, status, generation, clear_picture)
                }
                Err(_) => return glib::ControlFlow::Continue,
            };

            if clear_picture {
                picture.set_paintable(Option::<&gdk::Paintable>::None);
            }
            if let Some(frame) = frame {
                let bytes = glib::Bytes::from_owned(frame.rgba);
                let texture = gdk::MemoryTexture::new(
                    frame.width,
                    frame.height,
                    gdk::MemoryFormat::R8g8b8a8,
                    &bytes,
                    frame.stride,
                );
                picture.set_paintable(Some(&texture));
            }
            if generation != last_generation {
                status_label.set_text(&status);
                last_generation = generation;
            }
            glib::ControlFlow::Continue
        });
        PreviewBinding { enabled, alive }
    }
}

#[cfg(not(coverage))]
struct PreviewFrame {
    width: i32,
    height: i32,
    stride: usize,
    rgba: Vec<u8>,
}

#[cfg(not(coverage))]
fn run_preview_feed(
    server_addr: Arc<Mutex<String>>,
    monitor_id: u32,
    active: Arc<AtomicBool>,
    shared: Arc<Mutex<SharedPreviewState>>,
) -> Result<()> {
    let (pipeline, appsrc, appsink) = build_preview_pipeline()?;
    pipeline
        .set_state(gst::State::Playing)
        .context("starting launcher preview pipeline")?;

    {
        let shared = Arc::clone(&shared);
        let appsink = appsink.clone();
        std::thread::spawn(move || loop {
            if let Some(sample) = appsink.try_pull_sample(gst::ClockTime::from_mseconds(250)) {
                if let Some(frame) = sample_to_frame(&sample) {
                    if let Ok(mut slot) = shared.lock() {
                        slot.push_frame(frame);
                    }
                }
            }
        });
    }

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("building preview tokio runtime")?;

    let _ = rt.block_on(async move {
        loop {
            if !active.load(Ordering::Relaxed) {
                set_shared_status(&shared, PREVIEW_IDLE_STATUS, true);
                tokio::time::sleep(Duration::from_millis(150)).await;
                continue;
            }

            set_shared_status(&shared, "Connecting relay preview...", true);
            let current_addr = match server_addr.lock() {
                Ok(value) => value.clone(),
                Err(_) => {
                    set_shared_status(&shared, "Preview address is unavailable.", true);
                    tokio::time::sleep(Duration::from_millis(750)).await;
                    continue;
                }
            };

            let channel = match Channel::from_shared(current_addr.clone()) {
                Ok(endpoint) => match endpoint.tcp_nodelay(true).connect().await {
                    Ok(channel) => channel,
                    Err(err) => {
                        warn!(monitor_id, ?err, "launcher preview connect failed");
                        set_shared_status(
                            &shared,
                            format!("Preview host is unavailable: {err}"),
                            true,
                        );
                        tokio::time::sleep(Duration::from_millis(750)).await;
                        continue;
                    }
                },
                Err(err) => {
                    warn!(monitor_id, ?err, "launcher preview endpoint invalid");
                    set_shared_status(&shared, format!("Preview address is invalid: {err}"), true);
                    tokio::time::sleep(Duration::from_millis(750)).await;
                    continue;
                }
            };

            let mut cli = RelayClient::new(channel);
            let req = MonitorRequest {
                id: monitor_id,
                max_bitrate: preview_max_bitrate(),
            };
            match cli.capture_video(Request::new(req)).await {
                Ok(mut stream) => {
                    debug!(monitor_id, "launcher preview connected");
                    set_shared_status(&shared, "Waiting for stream...", true);
                    loop {
                        if !active.load(Ordering::Relaxed) {
                            break;
                        }
                        match tokio::time::timeout(
                            Duration::from_millis(300),
                            stream.get_mut().message(),
                        )
                        .await
                        {
                            Ok(Ok(Some(pkt))) => push_preview_packet(&appsrc, pkt),
                            Ok(Ok(None)) => {
                                set_shared_status(&shared, "Preview stream ended.", true);
                                break;
                            }
                            Ok(Err(err)) => {
                                warn!(monitor_id, ?err, "launcher preview stream error");
                                set_shared_status(
                                    &shared,
                                    format!("Preview stream error: {err}"),
                                    true,
                                );
                                break;
                            }
                            Err(_) => continue,
                        }
                    }
                }
                Err(err) => {
                    warn!(monitor_id, ?err, "launcher preview rpc failed");
                    set_shared_status(&shared, format!("Preview RPC failed: {err}"), true);
                }
            }
            tokio::time::sleep(Duration::from_millis(750)).await;
        }
        #[allow(unreachable_code)]
        Ok::<(), anyhow::Error>(())
    });

    Ok(())
}

#[cfg(not(coverage))]
fn set_shared_status(
    shared: &Arc<Mutex<SharedPreviewState>>,
    status: impl Into<String>,
    clear: bool,
) {
    if let Ok(mut slot) = shared.lock() {
        slot.set_status(status, clear);
    }
}

#[cfg(not(coverage))]
fn build_preview_pipeline() -> Result<(gst::Pipeline, gst_app::AppSrc, gst_app::AppSink)> {
    let desc = format!(
        "appsrc name=src is-live=true format=time do-timestamp=true block=false ! \
         queue max-size-buffers=6 max-size-time=0 max-size-bytes=0 leaky=downstream ! \
         h264parse disable-passthrough=true ! avdec_h264 ! videoconvert ! videoscale ! \
         video/x-raw,format=RGBA,width={PREVIEW_WIDTH},height={PREVIEW_HEIGHT},pixel-aspect-ratio=1/1 ! \
         appsink name=sink emit-signals=false sync=false max-buffers=1 drop=true"
    );
    let pipeline = gst::parse::launch(&desc)?
        .downcast::<gst::Pipeline>()
        .expect("preview pipeline");

    let appsrc = pipeline
        .by_name("src")
        .context("missing preview appsrc")?
        .downcast::<gst_app::AppSrc>()
        .expect("preview appsrc");
    appsrc.set_caps(Some(
        &gst::Caps::builder("video/x-h264")
            .field("stream-format", &"byte-stream")
            .field("alignment", &"au")
            .build(),
    ));
    appsrc.set_format(gst::Format::Time);

    let appsink = pipeline
        .by_name("sink")
        .context("missing preview appsink")?
        .downcast::<gst_app::AppSink>()
        .expect("preview appsink");
    appsink.set_caps(Some(
        &gst::Caps::builder("video/x-raw")
            .field("format", &"RGBA")
            .field("width", &PREVIEW_WIDTH)
            .field("height", &PREVIEW_HEIGHT)
            .build(),
    ));

    Ok((pipeline, appsrc, appsink))
}

#[cfg(not(coverage))]
fn push_preview_packet(appsrc: &gst_app::AppSrc, pkt: VideoPacket) {
    let mut buf = gst::Buffer::from_slice(pkt.data);
    if let Some(buf) = buf.get_mut() {
        buf.set_pts(Some(gst::ClockTime::from_useconds(pkt.pts)));
    }
    let _ = appsrc.push_buffer(buf);
}

#[cfg(not(coverage))]
fn sample_to_frame(sample: &gst::Sample) -> Option<PreviewFrame> {
    let caps = sample.caps()?;
    let structure = caps.structure(0)?;
    let width = structure.get::<i32>("width").ok()?;
    let height = structure.get::<i32>("height").ok()?;
    let buffer = sample.buffer()?;
    let map = buffer.map_readable().ok()?;
    let rgba = map.as_slice().to_vec();
    let stride = rgba.len() / height.max(1) as usize;
    Some(PreviewFrame {
        width,
        height,
        stride,
        rgba,
    })
}

#[cfg(not(coverage))]
fn preview_max_bitrate() -> u32 {
    std::env::var("LESAVKA_PREVIEW_MAX_KBIT")
        .ok()
        .and_then(|raw| raw.parse::<u32>().ok())
        .unwrap_or(2_500)
}