Take the current running time as PTS for now and clean up timestamp/duration calculations

We might want to take the timestamp provided by the NDI SDK at a later
time but this needs to be done correctly.

src/lib.rs

@@ -41,17 +41,8 @@ struct ReceiverInfo {
     video: bool,
     audio: bool,
     ndi_instance: RecvInstance,
-    initial_timestamp: u64,
 }
 
-struct Ndi {
-    start_pts: gst::ClockTime,
-}
-
-static mut NDI_STRUCT: Ndi = Ndi {
-    start_pts: gst::ClockTime(Some(0)),
-};
-
 lazy_static! {
     static ref HASHMAP_RECEIVERS: Mutex<HashMap<usize, ReceiverInfo>> = {
         let m = HashMap::new();
@@ -167,7 +158,6 @@ fn connect_ndi(
             video,
             audio: !video,
             ndi_instance: recv,
-            initial_timestamp: 0,
             id: id_receiver,
         },
     );

src/ndiaudiosrc.rs

@@ -16,7 +16,6 @@ use connect_ndi;
 use stop_ndi;
 use ndi::*;
 
-use NDI_STRUCT;
 use HASHMAP_RECEIVERS;
 
 use byte_slice_cast::AsMutSliceOf;
@@ -86,15 +85,10 @@ impl Default for State {
     }
 }
 
-struct TimestampData {
-    offset: u64,
-}
-
 pub(crate) struct NdiAudioSrc {
     cat: gst::DebugCategory,
     settings: Mutex<Settings>,
     state: Mutex<State>,
-    timestamp_data: Mutex<TimestampData>,
 }
 
 impl ObjectSubclass for NdiAudioSrc {
@@ -114,7 +108,6 @@ impl ObjectSubclass for NdiAudioSrc {
             ),
             settings: Mutex::new(Default::default()),
             state: Mutex::new(Default::default()),
-            timestamp_data: Mutex::new(TimestampData { offset: 0 }),
         }
     }
 
@@ -241,52 +234,6 @@ impl ObjectImpl for NdiAudioSrc {
 }
 
 impl ElementImpl for NdiAudioSrc {
-    fn change_state(
-        &self,
-        element: &gst::Element,
-        transition: gst::StateChange,
-    ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
-        if transition == gst::StateChange::PausedToPlaying {
-            let mut receivers = HASHMAP_RECEIVERS.lock().unwrap();
-            let state = self.state.lock().unwrap();
-
-            let receiver = receivers.get_mut(&state.id_receiver.unwrap()).unwrap();
-            let recv = &receiver.ndi_instance;
-
-            // FIXME error handling, make interruptable
-            let audio_frame =
-            loop {
-                match recv.capture(false, true, false, 1000) {
-                    Err(_) => unimplemented!(),
-                    Ok(None) => continue,
-                    Ok(Some(Frame::Audio(frame))) => break frame,
-                    _ => unreachable!(),
-                }
-            };
-
-            gst_debug!(
-                self.cat,
-                obj: element,
-                "NDI audio frame received: {:?}",
-                audio_frame
-            );
-
-            // FIXME handle unset timestamp
-            if receiver.initial_timestamp <= audio_frame.timestamp() as u64
-                || receiver.initial_timestamp == 0
-            {
-                receiver.initial_timestamp = audio_frame.timestamp() as u64;
-            }
-
-            gst_debug!(
-                self.cat,
-                obj: element,
-                "Setting initial timestamp to {}",
-                receiver.initial_timestamp
-            );
-        }
-        self.parent_change_state(element, transition)
-    }
 }
 
 impl BaseSrcImpl for NdiAudioSrc {
@@ -420,10 +367,8 @@ impl BaseSrcImpl for NdiAudioSrc {
         _offset: u64,
         _length: u32,
     ) -> Result<gst::Buffer, gst::FlowError> {
+        // FIXME: Make sure to not have any mutexes locked while wait
         let settings = self.settings.lock().unwrap().clone();
-
-        let mut timestamp_data = self.timestamp_data.lock().unwrap();
-
         let state = self.state.lock().unwrap();
         let _info = match state.info {
             None => {
@@ -437,7 +382,7 @@ impl BaseSrcImpl for NdiAudioSrc {
         let receiver = &receivers.get(&state.id_receiver.unwrap()).unwrap();
         let recv = &receiver.ndi_instance;
 
-        let time = receiver.initial_timestamp;
+        let clock = element.get_clock().unwrap();
 
         let mut count_frame_none = 0;
         let audio_frame =
@@ -478,13 +423,15 @@ impl BaseSrcImpl for NdiAudioSrc {
                 Ok(Some(frame)) => frame,
             };
 
-            if time >= (audio_frame.timestamp() as u64) {
+            break audio_frame;
-                gst_debug!(self.cat, obj: element, "Frame timestamp ({:?}) is lower than received in the first frame from NDI ({:?}), so skiping...", (audio_frame.timestamp() as u64), time);
-            } else {
-                break audio_frame;
-            }
         };
 
+        // For now take the current running time as PTS. At a later time we
+        // will want to work with the timestamp given by the NDI SDK if available
+        let now = clock.get_time();
+        let base_time = element.get_base_time();
+        let pts = now - base_time;
+
         gst_log!(
             self.cat,
             obj: element,
@@ -492,8 +439,6 @@ impl BaseSrcImpl for NdiAudioSrc {
             audio_frame
         );
 
-        let pts = audio_frame.timestamp() as u64 - time;
-
         gst_log!(
             self.cat,
             obj: element,
@@ -505,30 +450,12 @@ impl BaseSrcImpl for NdiAudioSrc {
         let buff_size = (audio_frame.no_samples() * 2 * audio_frame.no_channels()) as usize;
         let mut buffer = gst::Buffer::with_size(buff_size).unwrap();
         {
-            // FIXME don't use static mut, also this calculation is wrong
+            let duration = gst::SECOND.mul_div_floor(audio_frame.no_samples() as u64, audio_frame.sample_rate() as u64).unwrap_or(gst::CLOCK_TIME_NONE);
-            unsafe {
-                if NDI_STRUCT.start_pts == gst::ClockTime(Some(0)) {
-                    NDI_STRUCT.start_pts =
-                        element.get_clock().unwrap().get_time() - element.get_base_time();
-                }
-            }
-
             let buffer = buffer.get_mut().unwrap();
 
-            // Newtek NDI yields times in 100ns intervals since the Unix Time
+            buffer.set_pts(pts);
-            let pts: gst::ClockTime = (pts * 100).into();
-            buffer.set_pts(pts + unsafe { NDI_STRUCT.start_pts });
-
-            let duration: gst::ClockTime = (((f64::from(audio_frame.no_samples())
-                / f64::from(audio_frame.sample_rate()))
-                * 1_000_000_000.0) as u64)
-                .into();
             buffer.set_duration(duration);
 
-            buffer.set_offset(timestamp_data.offset);
-            timestamp_data.offset += audio_frame.no_samples() as u64;
-            buffer.set_offset_end(timestamp_data.offset);
-
             audio_frame.copy_to_interleaved_16s(buffer
                 .map_writable()
                 .unwrap()

src/ndivideosrc.rs

@@ -19,7 +19,6 @@ use ndi::*;
 use connect_ndi;
 use stop_ndi;
 
-use NDI_STRUCT;
 use HASHMAP_RECEIVERS;
 
 #[derive(Debug, Clone)]
@@ -87,15 +86,10 @@ impl Default for State {
     }
 }
 
-struct TimestampData {
-    offset: u64,
-}
-
 pub(crate) struct NdiVideoSrc {
     cat: gst::DebugCategory,
     settings: Mutex<Settings>,
     state: Mutex<State>,
-    timestamp_data: Mutex<TimestampData>,
 }
 
 impl ObjectSubclass for NdiVideoSrc {
@@ -115,7 +109,6 @@ impl ObjectSubclass for NdiVideoSrc {
             ),
             settings: Mutex::new(Default::default()),
             state: Mutex::new(Default::default()),
-            timestamp_data: Mutex::new(TimestampData { offset: 0 }),
         }
     }
 
@@ -249,52 +242,6 @@ impl ObjectImpl for NdiVideoSrc {
 }
 
 impl ElementImpl for NdiVideoSrc {
-    fn change_state(
-        &self,
-        element: &gst::Element,
-        transition: gst::StateChange,
-    ) -> Result<gst::StateChangeSuccess, gst::StateChangeError> {
-        if transition == gst::StateChange::PausedToPlaying {
-            let mut receivers = HASHMAP_RECEIVERS.lock().unwrap();
-            let state = self.state.lock().unwrap();
-
-            let receiver = receivers.get_mut(&state.id_receiver.unwrap()).unwrap();
-            let recv = &receiver.ndi_instance;
-
-            // FIXME error handling, make interruptable
-            let video_frame =
-            loop {
-                match recv.capture(true, false, false, 1000) {
-                    Err(_) => unimplemented!(),
-                    Ok(None) => continue,
-                    Ok(Some(Frame::Video(frame))) => break frame,
-                    _ => unreachable!(),
-                }
-            };
-
-            gst_debug!(
-                self.cat,
-                obj: element,
-                "NDI video frame received: {:?}",
-                video_frame
-            );
-
-            // FIXME handle unset timestamp
-
-            if receiver.initial_timestamp <= video_frame.timestamp() as u64
-                || receiver.initial_timestamp == 0
-            {
-                receiver.initial_timestamp = video_frame.timestamp() as u64;
-            }
-            gst_debug!(
-                self.cat,
-                obj: element,
-                "Setting initial timestamp to {}",
-                receiver.initial_timestamp
-            );
-        }
-        self.parent_change_state(element, transition)
-    }
 }
 
 impl BaseSrcImpl for NdiVideoSrc {
@@ -421,7 +368,7 @@ impl BaseSrcImpl for NdiVideoSrc {
         _offset: u64,
         _length: u32,
     ) -> Result<gst::Buffer, gst::FlowError> {
-        let mut timestamp_data = self.timestamp_data.lock().unwrap();
+        // FIXME: Make sure to not have any mutexes locked while wait
         let settings = self.settings.lock().unwrap().clone();
         let state = self.state.lock().unwrap();
         let _info = match state.info {
@@ -436,7 +383,7 @@ impl BaseSrcImpl for NdiVideoSrc {
         let receiver = &receivers.get(&state.id_receiver.unwrap()).unwrap();
         let recv = &receiver.ndi_instance;
 
-        let time = receiver.initial_timestamp;
+        let clock = element.get_clock().unwrap();
 
         let mut count_frame_none = 0;
         let video_frame =
@@ -477,13 +424,15 @@ impl BaseSrcImpl for NdiVideoSrc {
                 Ok(Some(frame)) => frame,
             };
 
-            if time >= (video_frame.timestamp() as u64) {
+            break video_frame;
-                gst_debug!(self.cat, obj: element, "Frame timestamp ({:?}) is lower than received in the first frame from NDI ({:?}), so skiping...", (video_frame.timestamp() as u64), time);
-            } else {
-                break video_frame;
-            }
         };
 
+        // For now take the current running time as PTS. At a later time we
+        // will want to work with the timestamp given by the NDI SDK if available
+        let now = clock.get_time();
+        let base_time = element.get_base_time();
+        let pts = now - base_time;
+
         gst_log!(
             self.cat,
             obj: element,
@@ -491,8 +440,6 @@ impl BaseSrcImpl for NdiVideoSrc {
             video_frame
         );
 
-        let pts = video_frame.timestamp() as u64 - time;
-
         gst_log!(
             self.cat,
             obj: element,
@@ -503,28 +450,10 @@ impl BaseSrcImpl for NdiVideoSrc {
         let buff_size = (video_frame.yres() * video_frame.line_stride_in_bytes()) as usize;
         let mut buffer = gst::Buffer::with_size(buff_size).unwrap();
         {
-            // Newtek NDI yields times in 100ns intervals since the Unix Time
+            let duration = gst::SECOND.mul_div_floor(video_frame.frame_rate().1 as u64, video_frame.frame_rate().0 as u64).unwrap_or(gst::CLOCK_TIME_NONE);
-            let pts: gst::ClockTime = (pts * 100).into();
-
-            let duration: gst::ClockTime = (((f64::from(video_frame.frame_rate().1)
-                / f64::from(video_frame.frame_rate().0))
-                * 1_000_000_000.0) as u64)
-                .into();
             let buffer = buffer.get_mut().unwrap();
-
+            buffer.set_pts(pts);
-            // FIXME don't use static mut, also this calculation is wrong
-            unsafe {
-                if NDI_STRUCT.start_pts == gst::ClockTime(Some(0)) {
-                    NDI_STRUCT.start_pts =
-                        element.get_clock().unwrap().get_time() - element.get_base_time();
-                }
-
-                buffer.set_pts(pts + NDI_STRUCT.start_pts);
-            }
             buffer.set_duration(duration);
-            buffer.set_offset(timestamp_data.offset);
-            timestamp_data.offset += 1;
-            buffer.set_offset_end(timestamp_data.offset);
             buffer.copy_from_slice(0, video_frame.data()).unwrap();
         }