livesync: accept variable framerate with fix downstream rate

The livesync element is produces a constant stream with max-framerate if
the upstream element negotiated variable framerate.

Thus, negotiation with downstream elements that request a fixed
framerate that is max-framerate should not fail.

utils/livesync/src/livesync/imp.rs

@@ -753,6 +753,118 @@ impl LiveSync {
         }
     }
 
+    fn transform_caps(
+        &self,
+        direction: gst::PadDirection,
+        caps: &gst::Caps,
+        filter: Option<&gst::Caps>,
+    ) -> Option<gst::Caps> {
+        let other_caps = if direction == gst::PadDirection::Src {
+            let mut caps = caps.clone();
+            let mut x = gst::Caps::new_empty();
+
+            for s in caps.make_mut().iter_mut() {
+                if let Ok(Some(framerate)) = s.get_optional::<gst::Fraction>("framerate") {
+                    if framerate.numer() != 0 {
+                        let mut f = s.to_owned();
+                        f.set("framerate", gst::Fraction::new(0, 1));
+                        f.set("max-framerate", framerate);
+                        x.merge_structure(f);
+                    }
+                }
+            }
+
+            caps.merge(x);
+            caps
+        } else {
+            let mut caps = caps.clone();
+            let mut x = gst::Caps::new_empty();
+
+            for s in caps.make_mut().iter_mut() {
+                if let Ok(Some(framerate)) = s.get_optional::<gst::Fraction>("framerate") {
+                    if framerate.numer() == 0 && framerate.denom() == 1 {
+                        let mut f = s.to_owned();
+                        if let Ok(Some(r)) = f.get_optional::<gst::Fraction>("max-framerate") {
+                            f.set("framerate", r);
+                            f.remove_field("max-framerate");
+                        } else {
+                            f.remove_field("framerate");
+                        }
+                        x.merge_structure(f);
+                    }
+                }
+            }
+
+            caps.merge(x);
+            caps
+        };
+
+        gst::debug!(
+            CAT,
+            imp: self,
+            "Transformed caps from {} to {} in direction {:?}",
+            caps,
+            other_caps,
+            direction
+        );
+
+        if let Some(filter) = filter {
+            Some(filter.intersect_with_mode(&other_caps, gst::CapsIntersectMode::First))
+        } else {
+            Some(other_caps)
+        }
+    }
+
+    fn find_transform(&self, pad: &gst::Pad, caps: &gst::Caps) -> Option<gst::Caps> {
+        let otherpad = if *pad == self.srcpad {
+            &self.sinkpad
+        } else {
+            &self.srcpad
+        };
+
+        let othercaps = self.transform_caps(pad.direction(), caps, None);
+
+        let peercaps = otherpad.peer_query_caps(othercaps.as_ref());
+
+        let mut othercaps = if let Some(c) = othercaps {
+            peercaps.intersect_with_mode(&c, gst::CapsIntersectMode::First)
+        } else {
+            peercaps
+        };
+        othercaps.fixate();
+
+        gst::debug!(CAT, imp: self, "Input caps were {:?}, and got final caps {:?}", caps, othercaps);
+
+        Some(othercaps)
+    }
+
+    fn accept_caps(&self, direction: gst::PadDirection, caps: &gst::Caps) -> bool {
+        let other_caps = self.transform_caps(direction, caps, None);
+        if let Some(o) = other_caps {
+            !o.is_empty()
+        } else {
+            true
+        }
+    }
+
+    fn query_caps(&self, pad: &gst::Pad, filter: Option<&gst::Caps>) -> Option<gst::Caps> {
+        let otherpad = if *pad == self.srcpad {
+            &self.sinkpad
+        } else {
+            &self.srcpad
+        };
+
+        let peerfilter = if let Some(f) = filter {
+            self.transform_caps(pad.direction(), f, None)
+        } else {
+            None
+        };
+
+        let peercaps = otherpad.peer_query_caps(peerfilter.as_ref());
+
+        self.transform_caps(otherpad.direction(), &peercaps, filter)
+    }
+
     fn sink_query(&self, pad: &gst::Pad, query: &mut gst::QueryRef) -> bool {
         if query.is_serialized() {
             let (sender, receiver) = mpsc::sync_channel(1);
@@ -772,7 +884,21 @@ impl LiveSync {
             // If the sender gets dropped, we will also unblock
             receiver.recv().unwrap_or(false)
         } else {
-            gst::Pad::query_default(pad, Some(&*self.obj()), query)
+            match query.view_mut() {
+                gst::QueryViewMut::Caps(q) => {
+                    let caps = self.query_caps(pad, q.filter_owned().as_ref());
+                    q.set_result(&caps);
+                    true
+                }
+                gst::QueryViewMut::AcceptCaps(q) => {
+                    let ret = self.accept_caps(gst::PadDirection::Src, &q.caps_owned());
+
+                    q.set_result(ret);
+
+                    true
+                }
+                _ => gst::Pad::query_default(pad, Some(&*self.obj()), query),
+            }
         }
     }
 
@@ -798,6 +924,12 @@ impl LiveSync {
                 true
             }
 
+            gst::QueryViewMut::Caps(q) => {
+                let caps = self.query_caps(pad, q.filter_owned().as_ref());
+                q.set_result(&caps);
+                true
+            }
+
             _ => gst::Pad::query_default(pad, Some(&*self.obj()), query),
         }
     }
@@ -1074,7 +1206,7 @@ impl LiveSync {
                     }
 
                     gst::EventView::Caps(e) => {
-                        state.pending_caps = Some(e.caps_owned());
+                        state.pending_caps = self.find_transform(&self.sinkpad, &e.caps_owned());
                         push = false;
                     }
 

utils/livesync/tests/livesync.rs

@@ -146,6 +146,37 @@ fn test_video_negotiate_framerate_variable() {
     assert_eq!(current_caps, src_caps);
 }
 
+#[test]
+fn test_video_negotiate_framerate_downstream() {
+    init();
+
+    let mut h = gst_check::Harness::new("livesync");
+
+    let sink_caps = gst::Caps::builder("video/x-raw")
+        .field("framerate", gst::Fraction::new(60, 1))
+        .build();
+    h.set_sink_caps(sink_caps.clone());
+
+    let src_caps = gst::Caps::builder("video/x-raw")
+        .field("framerate", gst::Fraction::new(0, 1))
+        .field("max-framerate", gst::Fraction::new(60, 1))
+        .build();
+    h.set_src_caps(src_caps.clone());
+
+    h.play();
+
+    let buffer = gst::Buffer::new();
+    assert!(h.push_and_pull(buffer).is_ok());
+
+    let current_caps = h
+        .sinkpad()
+        .expect("harness has no sinkpad")
+        .current_caps()
+        .expect("current caps missing");
+
+    assert_eq!(current_caps, sink_caps);
+}
+
 fn test_video(singlesegment: bool) {
     init();