add capture time metadata

Author: lherman-cs

pulsebeam-agent/src/actor.rs

@@ -307,7 +307,7 @@ impl AgentActor {
                 Some((mid, frame)) = self.senders.next() => {
                      if let Some(writer) = self.rtc.writer(mid) {
                          let pt = writer.payload_params().nth(0).unwrap().pt();
-                         let _ = writer.write(pt, Instant::now().into(), frame.ts, frame.data);
+                         let _ = writer.write(pt, frame.capture_time.into(), frame.ts, frame.data);
                      }
                 }
             }
@@ -321,10 +321,7 @@ impl AgentActor {
             Event::MediaData(data) => {
                 // Network -> User
                 if let Some(tx) = self.receivers.get(&data.mid) {
-                    let _ = tx.try_send(MediaFrame {
-                        data: Bytes::from(data.data),
-                        ts: data.time,
-                    });
+                    let _ = tx.try_send(data.into());
                 }
             }
 

pulsebeam-agent/src/lib.rs

@@ -1,22 +1,24 @@
 pub use bytes::Bytes;
 pub use str0m::Candidate;
 pub use str0m::media::{MediaData, MediaKind, MediaTime};
+use tokio::time::Instant;
 
 pub mod actor;
 pub mod media;
 pub mod signaling;
-// pub mod rt;
 
 pub struct MediaFrame {
     pub ts: MediaTime,
     pub data: Bytes,
+    pub capture_time: Instant,
 }
 
 impl From<MediaData> for MediaFrame {
     fn from(value: MediaData) -> Self {
         Self {
             ts: value.time,
             data: value.data.into(),
+            capture_time: value.network_time.into(),
         }
     }
 }

pulsebeam-cli/src/main.rs

@@ -5,7 +5,6 @@ use pulsebeam_agent::{
     signaling::HttpSignalingClient,
 };
 use std::time::Duration;
-use tokio::time::Instant;
 
 const RAW_H264: &[u8] = include_bytes!("video.h264");
 
@@ -33,20 +32,21 @@ async fn main() {
 fn handle_sender(sender: TrackSender) {
     tokio::spawn(async move {
         let mut looper = H264Looper::new(RAW_H264);
-        let mut interval = tokio::time::interval(Duration::from_nanos(1_000_000_000 / 30));
-        let start = Instant::now();
-        loop {
-            let now = interval.tick().await;
-            let elapsed = now - start;
-
-            // Calculate TS based on actual elapsed time to prevent drift
-            let ts = (elapsed.as_secs_f64() * 90000.0) as u64;
+        let frame_duration_90khz = 90000 / 30;
+        let mut ticker = tokio::time::interval(Duration::from_nanos(1_000_000_000 / 30));
+        let mut current_ts: u64 = 0;
 
+        loop {
+            let now = ticker.tick().await;
+            let frame_data = looper.next().unwrap();
             let frame = MediaFrame {
-                ts: MediaTime::from_90khz(ts),
-                data: looper.next().unwrap(),
+                ts: MediaTime::from_90khz(current_ts),
+                data: frame_data,
+                capture_time: now,
             };
+
             sender.try_send(frame);
+            current_ts += frame_duration_90khz;
         }
     });
 }