video: pause encoding while nobody is watching

app/src/pipeline.rs

@@ -87,6 +87,7 @@ pub struct VideoPipeline {
     pipeline: gst::Pipeline,
     appsink: gst_app::AppSink,
     encoder: gst::Element,
+    valve: gst::Element,
     running: Arc<AtomicBool>,
     raw_frame_waiters: Arc<RawFrameWaiters>,
     raw_frame_requested: Arc<AtomicBool>,
@@ -145,6 +146,17 @@ impl VideoPipeline {
             .build()
             .context("Failed to create queue")?;
 
+        // Gates the encoder: while nobody is watching it drops raw frames so the
+        // hardware encoder and its output thread sit idle. Sticky events (caps)
+        // still pass, so the encoder stays negotiated and video state, dynamic
+        // GOP and the screenshot probe keep working. Starts closed.
+        let valve = gst::ElementFactory::make("valve")
+            .name("encode_valve")
+            .property_from_str("drop-mode", "forward-sticky-events")
+            .property("drop", true)
+            .build()
+            .context("Failed to create valve")?;
+
         let encoder_name = config.encoder.gst_element_name();
         info!("Creating {} encoder for {}", encoder_name, config.encoder.codec_name());
 
@@ -190,6 +202,7 @@ impl VideoPipeline {
                 &videorate,
                 &rate_caps,
                 &queue,
+                &valve,
                 &encoder,
                 &parser,
                 appsink.upcast_ref(),
@@ -203,6 +216,7 @@ impl VideoPipeline {
             &videorate,
             &rate_caps,
             &queue,
+            &valve,
             &encoder,
             &parser,
             appsink.upcast_ref(),
@@ -215,7 +229,8 @@ impl VideoPipeline {
         // steady-state cost per frame is a single relaxed atomic load.
         let probe_waiters = raw_frame_waiters.clone();
         let probe_requested = raw_frame_requested.clone();
-        encoder.static_pad("sink").context("Encoder has no sink pad")?.add_probe(
+        // On the valve's input, so snapshots work while encoding is paused.
+        valve.static_pad("sink").context("Valve has no sink pad")?.add_probe(
             gst::PadProbeType::BUFFER,
             move |pad, info| {
                 if !probe_requested.swap(false, Ordering::AcqRel) {
@@ -268,6 +283,7 @@ impl VideoPipeline {
             pipeline,
             appsink,
             encoder: encoder_for_state,
+            valve,
             running,
             raw_frame_waiters,
             raw_frame_requested,
@@ -302,6 +318,17 @@ impl VideoPipeline {
         self.running.load(Ordering::SeqCst)
     }
 
+    /// Opens or closes the encoder valve. Returns whether this changed it.
+    pub fn set_encoding_active(&self, active: bool) -> bool {
+        let drop = !active;
+        if self.valve.property::<bool>("drop") == drop {
+            return false;
+        }
+        self.valve.set_property("drop", drop);
+        info!(active, "video encoding {}", if active { "resumed" } else { "paused" });
+        true
+    }
+
     pub fn stop(&self) -> Result<()> {
         self.running.store(false, Ordering::SeqCst);
 
@@ -786,6 +813,14 @@ impl PipelineManager {
             warn!("force-keyframe event was not handled by encoder");
         }
     }
+
+    /// Resumes encoding for a viewer (with a keyframe so it can decode at once)
+    /// or pauses it when nobody is watching.
+    pub fn set_encoding_active(&self, active: bool) {
+        if self.video.set_encoding_active(active) && active {
+            self.force_keyframe();
+        }
+    }
 }
 
 pub fn encode_jpeg(sample: &gst::Sample) -> Result<bytes::Bytes> {

app/src/video.rs

@@ -158,6 +158,14 @@ pub async fn attach_audio_sink(track: Arc<TrackLocalStaticSample>) {
 pub async fn detach_webrtc_sink() {
     *VIDEO_SINK.write().await = None;
     *AUDIO_SINK.write().await = None;
+    set_encoding_active(false).await;
+}
+
+/// Encoding only runs while a session has a video track attached.
+pub async fn set_encoding_active(active: bool) {
+    if let Some(manager) = PIPELINE_MANAGER.lock().await.as_ref() {
+        manager.set_encoding_active(active);
+    }
 }
 
 async fn audio_frame_writer(mut rx: mpsc::Receiver<AudioFrame>) {
@@ -252,6 +260,8 @@ async fn start_pipeline(
     }
 
     manager.start()?;
+    // A restarted pipeline comes up paused; resume it if a viewer is attached.
+    manager.set_encoding_active(VIDEO_SINK.read().await.is_some());
 
     *PIPELINE_MANAGER.lock().await = Some(manager);
     Ok(())

app/src/webrtc.rs

@@ -258,6 +258,7 @@ impl WebRTCApi {
                                 increment_session_counter().await;
                                 if let Some(track) = video_track.clone() {
                                     video::attach_webrtc_sink(track).await;
+                                    video::set_encoding_active(true).await;
                                     info!("Video track attached to WebRTC session {}", session_id);
                                 }
                                 if let Some(track) = audio_track.clone() {

docs/api.md

@@ -213,7 +213,7 @@ enabled the device announces itself under `homeassistant/…/rustkvm_<device id>
 | Metric | Type |
 | --- | --- |
 | `rustkvm_app_info` | gauge with build labels |
-| `rustkvm_video_frames_total` | counter |
+| `rustkvm_video_frames_total` | counter (encoded frames; encoding pauses while no session is watching) |
 | `rustkvm_audio_frames_total` | counter |
 | `rustkvm_video_pipeline_restarts_total` | counter |
 | `rustkvm_rpc_calls_total{method}` | counter |