Fix ClockTime comparisons not being Ord and use saturating_sub

See:
https://gitlab.freedesktop.org/gstreamer/gstreamer-rs/-/merge_requests/607

4 files changed, 27 insertions, 26 deletions

diff --git a/generic/threadshare/src/inputselector.rs b/generic/threadshare/src/inputselector.rs
index cfb9bd260..295d4537e 100644
--- a/generic/threadshare/src/inputselector.rs
+++ b/generic/threadshare/src/inputselector.rs
@@ -113,9 +113,11 @@ impl InputSelectorPadSinkHandler {
     async fn sync(&self, element: &gst::Element, running_time: gst::ClockTime) {
         let now = element.get_current_running_time();
 
-        if now.is_some() && now < running_time {
-            let delay = running_time - now;
-            runtime::time::delay_for(Duration::from_nanos(delay.nseconds().unwrap())).await;
+        if let Some(delay) = running_time
+            .saturating_sub(now)
+            .and_then(|delay| delay.nseconds())
+        {
+            runtime::time::delay_for(Duration::from_nanos(delay)).await;
         }
     }
 
@@ -324,7 +326,7 @@ impl PadSrcHandler for InputSelectorPadSrcHandler {
             QueryView::Latency(ref mut q) => {
                 let mut ret = true;
                 let mut min_latency = 0.into();
-                let mut max_latency = gst::CLOCK_TIME_NONE;
+                let mut max_latency = gst::ClockTime::none();
                 let pads = {
                     let pads = inputselector.pads.lock().unwrap();
                     pads.sink_pads
@@ -341,12 +343,8 @@ impl PadSrcHandler for InputSelectorPadSrcHandler {
                     if ret {
                         let (live, min, max) = peer_query.get_result();
                         if live {
-                            min_latency = std::cmp::max(min, min_latency);
-                            if max_latency.is_none() && max.is_some() {
-                                max_latency = max;
-                            } else if max_latency.is_some() && max.is_some() {
-                                max_latency = std::cmp::min(max, max_latency);
-                            }
+                            min_latency = min.max(min_latency).unwrap_or(min_latency);
+                            max_latency = max.min(max_latency).unwrap_or(max);
                         }
                     }
                 }
diff --git a/generic/threadshare/src/jitterbuffer/jitterbuffer.rs b/generic/threadshare/src/jitterbuffer/jitterbuffer.rs
index 7191d90b5..3579f9f3a 100644
--- a/generic/threadshare/src/jitterbuffer/jitterbuffer.rs
+++ b/generic/threadshare/src/jitterbuffer/jitterbuffer.rs
@@ -295,9 +295,10 @@ impl SinkHandler {
                 let new_packet_spacing = pts - inner.ips_pts;
                 let old_packet_spacing = state.packet_spacing;
 
+                assert!(old_packet_spacing.is_some());
                 if old_packet_spacing > new_packet_spacing {
                     state.packet_spacing = (new_packet_spacing + 3 * old_packet_spacing) / 4;
-                } else if old_packet_spacing > gst::ClockTime(Some(0)) {
+                } else if !old_packet_spacing.is_zero() {
                     state.packet_spacing = (3 * new_packet_spacing + old_packet_spacing) / 4;
                 } else {
                     state.packet_spacing = new_packet_spacing;
@@ -412,13 +413,13 @@ impl SinkHandler {
             pt
         );
 
-        if dts == gst::CLOCK_TIME_NONE {
+        if dts.is_none() {
             dts = pts;
-        } else if pts == gst::CLOCK_TIME_NONE {
+        } else if pts.is_none() {
             pts = dts;
         }
 
-        if dts == gst::CLOCK_TIME_NONE {
+        if dts.is_none() {
             dts = element.get_current_running_time();
             pts = dts;
 
@@ -957,13 +958,11 @@ impl SrcHandler {
 
         let next_wakeup = state.earliest_pts + latency - state.packet_spacing - context_wait / 2;
 
-        let delay = {
-            if next_wakeup > now {
-                (next_wakeup - now).nseconds().unwrap()
-            } else {
-                0
-            }
-        };
+        let delay = next_wakeup
+            .saturating_sub(now)
+            .unwrap_or_else(gst::ClockTime::zero)
+            .nseconds()
+            .unwrap();
 
         gst_debug!(
             CAT,
@@ -1121,7 +1120,7 @@ impl Default for State {
             segment: gst::FormattedSegment::<gst::ClockTime>::new(),
             clock_rate: None,
 
-            packet_spacing: gst::ClockTime(Some(0)),
+            packet_spacing: gst::ClockTime::zero(),
             equidistant: 0,
 
             discont: true,
diff --git a/generic/threadshare/src/runtime/time.rs b/generic/threadshare/src/runtime/time.rs
index 0ea857b0c..5dac6c49a 100644
--- a/generic/threadshare/src/runtime/time.rs
+++ b/generic/threadshare/src/runtime/time.rs
@@ -26,7 +26,9 @@ use std::time::Duration;
 ///
 /// This must be called from within the target runtime environment.
 pub async fn delay_for(delay: Duration) {
-    tokio::time::delay_for(delay).map(drop).await;
+    if delay > Duration::from_nanos(0) {
+        tokio::time::delay_for(delay).map(drop).await;
+    }
 }
 
 /// Builds a `Stream` that yields at `interval.
diff --git a/generic/threadshare/src/udpsink.rs b/generic/threadshare/src/udpsink.rs
index 46fa1c23d..35ea4c599 100644
--- a/generic/threadshare/src/udpsink.rs
+++ b/generic/threadshare/src/udpsink.rs
@@ -694,9 +694,11 @@ impl UdpSinkPadHandler {
     async fn sync(&self, element: &gst::Element, running_time: gst::ClockTime) {
         let now = element.get_current_running_time();
 
-        if now < running_time {
-            let delay = running_time - now;
-            runtime::time::delay_for(Duration::from_nanos(delay.nseconds().unwrap())).await;
+        if let Some(delay) = running_time
+            .saturating_sub(now)
+            .and_then(|delay| delay.nseconds())
+        {
+            runtime::time::delay_for(Duration::from_nanos(delay)).await;
         }
     }