Restore decoupling

src/lib/stream/gst/utils.rs

@@ -685,39 +685,6 @@ pub fn excise_single_element(element: &gst::Element) -> Result<()> {
     Ok(())
 }
 
-/// Excises the internal queue that `proxysrc` creates, eliminating one frame of
-/// intermediate buffering. Must be called from a pad-probe on the queue's src pad
-/// once the first buffer has flowed.
-pub fn excise_proxysrc_queue(queue: &glib::WeakRef<gst::Element>) {
-    let Some(queue_ref) = queue.upgrade() else {
-        return;
-    };
-
-    let Some(upstream_pad) = queue_ref
-        .static_pad("sink")
-        .and_then(|sink_pad| sink_pad.peer())
-    else {
-        warn!("Failed to find upstream pad for proxysrc queue excision");
-        return;
-    };
-
-    info!("Proxysrc queue excision: Installing BLOCK_DOWNSTREAM probe");
-
-    let queue = queue.clone();
-    upstream_pad.add_probe(gst::PadProbeType::BLOCK_DOWNSTREAM, move |_pad, _info| {
-        let Some(queue_ref) = queue.upgrade() else {
-            return gst::PadProbeReturn::Remove;
-        };
-
-        match excise_single_element(&queue_ref) {
-            Ok(()) => info!("Proxysrc queue excision: Queue successfully excised"),
-            Err(error) => warn!("Failed to excise proxysrc queue: {error:?}"),
-        }
-
-        gst::PadProbeReturn::Remove
-    });
-}
-
 /// Hooks into the pipeline to excise every `rtpjitterbuffer` that `rtspsrc`'s
 /// internal `rtpbin` creates. Without the jitterbuffer, retransmission (NACK)
 /// cannot function, so `do-retransmission` is also forced to `false`.

src/lib/stream/manager.rs

@@ -675,7 +675,13 @@ impl Manager {
                 session_id,
             };
 
-            let sink = Sink::WebRTC(WebRTCSink::try_new(bind, sender)?);
+            let video_and_stream_information =
+                stream.video_and_stream_information.read().await.clone();
+            let sink = Sink::WebRTC(WebRTCSink::try_new(
+                bind,
+                sender,
+                &video_and_stream_information,
+            )?);
 
             let mut state_guard = stream.state.write().await;
 

src/lib/stream/mod.rs

@@ -37,7 +37,6 @@ use self::{
     gst::utils::wait_for_element_state,
     lifecycle::{LifecycleState, Phase},
     rtsp::{rtsp_scheme::RTSPScheme, rtsp_server::RTSPServer},
-    sink::SinkInterface,
 };
 
 pub struct Stream {
@@ -871,16 +870,6 @@ impl StreamState {
             .pipeline_runner
             .start()?;
 
-        // Start all the sinks
-        if let Some(pipeline) = stream.pipeline.as_mut() {
-            let pipeline_state = pipeline.inner_state_mut();
-            for sink in pipeline_state.sinks.values() {
-                if let Err(error) = sink.start() {
-                    warn!("Failed to start sink: {error:?}");
-                }
-            }
-        }
-
         Ok(stream)
     }
 }

src/lib/stream/pipeline/mod.rs

@@ -284,6 +284,11 @@ impl PipelineState {
             error!("Failed to syncronize children states. Reason: {error:?}");
         }
 
+        // Start the sink's own sub-pipeline runner
+        if let Err(error) = sink.start() {
+            warn!("Failed to start sink {sink_id}: {error:?}");
+        }
+
         self.sinks.insert(**sink_id, sink);
 
         Ok(())

src/lib/stream/pipeline/onvif_pipeline.rs

@@ -77,7 +77,6 @@ impl OnvifPipeline {
                         " ! tee name={raw_rtp_tee} allow-not-linked=true",
                         " {raw_rtp_tee}.",
                         " ! rtph264depay source-info=true",
-                        " ! queue leaky=downstream silent=true flush-on-eos=true max-size-buffers=1 max-size-bytes=0 max-size-time=0",
                         " ! h264parse config-interval=-1",
                         " ! capsfilter name={filter_name} caps=video/x-h264,stream-format=avc,alignment=au",
                         " ! tee name={video_tee_name} allow-not-linked=true",
@@ -102,7 +101,6 @@ impl OnvifPipeline {
                         " ! tee name={raw_rtp_tee} allow-not-linked=true",
                         " {raw_rtp_tee}.",
                         " ! rtph265depay source-info=true",
-                        " ! queue leaky=downstream silent=true flush-on-eos=true max-size-buffers=1 max-size-bytes=0 max-size-time=0",
                         " ! h265parse config-interval=-1",
                         " ! capsfilter name={filter_name} caps=video/x-h265,stream-format=byte-stream,alignment=au",
                         " ! tee name={video_tee_name} allow-not-linked=true",

src/lib/stream/pipeline/redirect_pipeline.rs

@@ -116,7 +116,6 @@ impl RedirectPipeline {
                         " ! tee name={raw_rtp_tee} allow-not-linked=true",
                         " {raw_rtp_tee}.",
                         " ! rtph264depay source-info=true",
-                        " ! queue leaky=downstream silent=true flush-on-eos=true max-size-buffers=1 max-size-bytes=0 max-size-time=0",
                         " ! h264parse config-interval=-1",
                         " ! capsfilter name={filter_name} caps=video/x-h264,stream-format=avc,alignment=au",
                         " ! tee name={video_tee_name} allow-not-linked=true",
@@ -138,7 +137,6 @@ impl RedirectPipeline {
                         " ! tee name={raw_rtp_tee} allow-not-linked=true",
                         " {raw_rtp_tee}.",
                         " ! rtph265depay source-info=true",
-                        " ! queue leaky=downstream silent=true flush-on-eos=true max-size-buffers=1 max-size-bytes=0 max-size-time=0",
                         " ! h265parse config-interval=-1",
                         " ! capsfilter name={filter_name} caps=video/x-h265,stream-format=byte-stream,alignment=au",
                         " ! tee name={video_tee_name} allow-not-linked=true",

src/lib/stream/sink/mod.rs

@@ -130,6 +130,31 @@ pub async fn create_zenoh_sink(
     ))
 }
 
+#[instrument(level = "debug")]
+pub fn make_proxy_bridge() -> Result<[gst::Element; 2]> {
+    let proxysink = gst::ElementFactory::make("proxysink").build()?;
+    let proxysrc = gst::ElementFactory::make("proxysrc")
+        .property("proxysink", &proxysink)
+        .build()?;
+
+    let queue = proxysrc
+        .downcast_ref::<gst::Bin>()
+        .and_then(|bin| {
+            bin.children()
+                .into_iter()
+                .find(|element| element.name().starts_with("queue"))
+        })
+        .context("Failed to find queue inside proxysrc")?;
+
+    queue.set_property_from_str("leaky", "downstream");
+    queue.set_property("flush-on-eos", true);
+    queue.set_property("max-size-buffers", 0u32);
+    queue.set_property("max-size-bytes", 0u32);
+    queue.set_property("max-size-time", gst::ClockTime::from_seconds(5).nseconds());
+
+    Ok([proxysink, proxysrc])
+}
+
 #[instrument(level = "debug", skip_all)]
 pub fn link_sink_to_tee(
     tee_src_pad: &gst::Pad,
@@ -210,7 +235,7 @@ pub fn force_sync_false_on_element_tree(element: &gst::Element) {
 }
 
 pub fn force_sync_false_on_element(element: &gst::Element) -> bool {
-    if element.find_property("sync").is_none() || element.static_pad("src").is_some() {
+    if element.find_property("sync").is_none() {
         return false;
     }
 

src/lib/stream/sink/udp_sink.rs

@@ -5,19 +5,17 @@ use gst::prelude::*;
 use tracing::*;
 
 use crate::{
-    stream::{gst::utils::excise_proxysrc_queue, pipeline::runner::PipelineRunner},
-    video_stream::types::VideoAndStreamInformation,
+    stream::pipeline::runner::PipelineRunner, video_stream::types::VideoAndStreamInformation,
 };
 
-use super::{SinkInterface, link_sink_to_tee, unlink_sink_from_tee};
+use super::{SinkInterface, link_sink_to_tee, make_proxy_bridge, unlink_sink_from_tee};
 
 #[derive(Debug)]
 pub struct UdpSink {
     sink_id: Arc<uuid::Uuid>,
     pipeline: gst::Pipeline,
     proxysink: gst::Element,
     _proxysrc: gst::Element,
-    proxysrc_queue: Option<gst::Element>,
     _udpsink: gst::Element,
     udpsink_sink_pad: gst::Pad,
     tee_src_pad: Option<gst::Pad>,
@@ -46,19 +44,6 @@ impl SinkInterface for UdpSink {
         let elements = &[&self.proxysink];
         link_sink_to_tee(tee_src_pad, pipeline, elements)?;
 
-        if let Some(queue) = &self.proxysrc_queue
-            && let Some(src_pad) = queue.static_pad("src")
-        {
-            let queue_weak = queue.downgrade();
-            src_pad.add_probe(
-                gst::PadProbeType::BUFFER | gst::PadProbeType::BUFFER_LIST,
-                move |_pad, _info| {
-                    excise_proxysrc_queue(&queue_weak);
-                    gst::PadProbeReturn::Remove
-                },
-            );
-        }
-
         Ok(())
     }
 
@@ -160,26 +145,7 @@ impl UdpSink {
         sink_id: Arc<uuid::Uuid>,
         video_and_stream_information: &VideoAndStreamInformation,
     ) -> Result<Self> {
-        let proxysink = gst::ElementFactory::make("proxysink").build()?;
-        let _proxysrc = gst::ElementFactory::make("proxysrc")
-            .property("proxysink", &proxysink)
-            .build()?;
-
-        let proxysrc_queue = _proxysrc.downcast_ref::<gst::Bin>().and_then(|bin| {
-            bin.children()
-                .into_iter()
-                .find(|element| element.name().starts_with("queue"))
-        });
-        if let Some(queue) = &proxysrc_queue {
-            queue.set_property_from_str("leaky", "downstream");
-            queue.set_property("silent", true);
-            queue.set_property("flush-on-eos", true);
-            queue.set_property("max-size-buffers", 1u32);
-            queue.set_property("max-size-bytes", 0u32);
-            queue.set_property("max-size-time", 0u64);
-        } else {
-            warn!("Failed to find queue inside proxysrc");
-        }
+        let [proxysink, _proxysrc] = make_proxy_bridge()?;
 
         let addresses = video_and_stream_information
             .stream_information
@@ -237,7 +203,6 @@ impl UdpSink {
             pipeline,
             proxysink,
             _proxysrc,
-            proxysrc_queue,
             _udpsink,
             udpsink_sink_pad,
             addresses,