add MessageInner::SenderRtp/Rtcp and ReceiverRtcp

yngrtc · yngrtc · commit 1c8a6d2824ee · 2026-02-22T22:06:24.000-08:00
diff --git a/src/media_stream/track_local/mod.rs b/src/media_stream/track_local/mod.rs
@@ -1,10 +1,32 @@
+use crate::error::Result;
 use crate::media_stream::Track;
+use crate::peer_connection::MessageInner;
+use crate::runtime::Sender;
+use rtc::rtp_transceiver::RTCRtpSenderId;
 use rtc::{rtcp, rtp};
 
 pub mod static_rtp;
 
+/// TrackLocalContext is the Context passed when a TrackLocal has been Binded/Unbinded from a PeerConnection, and used
+/// in Interceptors.
+#[derive(Clone)]
+pub struct TrackLocalContext {
+    pub(crate) sender_id: RTCRtpSenderId,
+    pub(crate) msg_tx: Sender<MessageInner>,
+}
+
 #[async_trait::async_trait]
 pub trait TrackLocal: Track {
-    async fn write_rtp(&self, packet: rtp::Packet) -> crate::error::Result<()>;
-    async fn write_rtcp(&self, packets: Vec<Box<dyn rtcp::Packet>>) -> crate::error::Result<()>;
+    /// bind should implement the way how the media data flows from the Track to the PeerConnection
+    /// This will be called internally after signaling is complete and the list of available
+    /// codecs has been determined
+    async fn bind(&self, ctx: &TrackLocalContext) -> Result<()>;
+
+    /// unbind should implement the teardown logic when the track is no longer needed. This happens
+    /// because a track has been stopped.
+    async fn unbind(&self) -> Result<()>;
+
+    async fn write_rtp(&self, packet: rtp::Packet) -> Result<()>;
+
+    async fn write_rtcp(&self, packets: Vec<Box<dyn rtcp::Packet>>) -> Result<()>;
 }
diff --git a/src/media_stream/track_local/static_rtp.rs b/src/media_stream/track_local/static_rtp.rs
@@ -1,19 +1,25 @@
-use crate::error::Result;
+use crate::error::{Error, Result};
 use crate::media_stream::Track;
-use crate::media_stream::track_local::TrackLocal;
+use crate::media_stream::track_local::{TrackLocal, TrackLocalContext};
+use crate::peer_connection::MessageInner;
+use crate::runtime::Mutex;
 use rtc::media_stream::MediaStreamTrack;
-use rtc::rtp::Packet;
+use rtc::{rtcp, rtp};
 
 /// TrackLocalStaticRTP  is a TrackLocal that has a pre-set codec and accepts RTP Packets.
 /// If you wish to send a media.Sample use TrackLocalStaticSample
-#[derive(Debug)]
+#[derive(Clone)]
 pub struct TrackLocalStaticRTP {
     track: MediaStreamTrack,
+    ctx: Mutex<Option<TrackLocalContext>>,
 }
 
 impl TrackLocalStaticRTP {
     pub fn new(track: MediaStreamTrack) -> Self {
-        Self { track }
+        Self {
+            track,
+            ctx: Mutex::new(None),
+        }
     }
 }
 
@@ -25,11 +31,37 @@ impl Track for TrackLocalStaticRTP {
 
 #[async_trait::async_trait]
 impl TrackLocal for TrackLocalStaticRTP {
-    async fn write_rtp(&self, _packet: Packet) -> Result<()> {
-        todo!()
+    async fn bind(&self, ctx: &TrackLocalContext) -> Result<()> {
+        let mut ctx_opt = self.ctx.lock().await;
+        *ctx_opt = Some(ctx.clone());
+        Ok(())
     }
 
-    async fn write_rtcp(&self, _packets: Vec<Box<dyn rtc::rtcp::Packet>>) -> Result<()> {
-        todo!()
+    async fn unbind(&self) -> Result<()> {
+        let mut ctx_opt = self.ctx.lock().await;
+        *ctx_opt = None;
+        Ok(())
+    }
+
+    async fn write_rtp(&self, packet: rtp::Packet) -> Result<()> {
+        let ctx_opt = self.ctx.lock().await;
+        if let Some(ctx) = &*ctx_opt {
+            ctx.msg_tx
+                .try_send(MessageInner::SenderRtp(ctx.sender_id, packet))
+                .map_err(|e| Error::Other(format!("{:?}", e)))
+        } else {
+            Err(Error::Other("track is not binding yet".to_string()))
+        }
+    }
+
+    async fn write_rtcp(&self, packets: Vec<Box<dyn rtcp::Packet>>) -> Result<()> {
+        let ctx_opt = self.ctx.lock().await;
+        if let Some(ctx) = &*ctx_opt {
+            ctx.msg_tx
+                .try_send(MessageInner::SenderRtcp(ctx.sender_id, packets))
+                .map_err(|e| Error::Other(format!("{:?}", e)))
+        } else {
+            Err(Error::Other("track is not binding yet".to_string()))
+        }
     }
 }
diff --git a/src/media_stream/track_remote/mod.rs b/src/media_stream/track_remote/mod.rs
@@ -1,5 +1,6 @@
 pub(crate) mod static_rtp;
 
+use crate::error::Result;
 use crate::media_stream::Track;
 use rtc::{rtcp, rtp};
 
@@ -19,6 +20,7 @@ pub enum TrackRemoteEvent {
 
 #[async_trait::async_trait]
 pub trait TrackRemote: Track {
-    async fn write_rtcp(&self, packets: Vec<Box<dyn rtcp::Packet>>) -> crate::error::Result<()>;
+    async fn write_rtcp(&self, packets: Vec<Box<dyn rtcp::Packet>>) -> Result<()>;
+
     async fn poll(&self) -> Option<TrackRemoteEvent>;
 }
diff --git a/src/media_stream/track_remote/static_rtp.rs b/src/media_stream/track_remote/static_rtp.rs
@@ -1,19 +1,32 @@
+use crate::error::{Error, Result};
 use crate::media_stream::Track;
 use crate::media_stream::track_remote::{TrackRemote, TrackRemoteEvent};
-use crate::runtime::{Mutex, Receiver};
+use crate::peer_connection::MessageInner;
+use crate::runtime::{Mutex, Receiver, Sender};
 use rtc::media_stream::MediaStreamTrack;
+use rtc::rtp_transceiver::RTCRtpReceiverId;
 
 /// TrackLocalStaticRTP  is a TrackLocal that has a pre-set codec and accepts RTP Packets.
 /// If you wish to send a media.Sample use TrackLocalStaticSample
+#[derive(Clone)]
 pub(crate) struct TrackRemoteStaticRTP {
     track: MediaStreamTrack,
+    receiver_id: RTCRtpReceiverId,
+    msg_tx: Sender<MessageInner>,
     evt_rx: Mutex<Receiver<TrackRemoteEvent>>,
 }
 
 impl TrackRemoteStaticRTP {
-    pub fn new(track: MediaStreamTrack, evt_rx: Receiver<TrackRemoteEvent>) -> Self {
+    pub fn new(
+        track: MediaStreamTrack,
+        receiver_id: RTCRtpReceiverId,
+        msg_tx: Sender<MessageInner>,
+        evt_rx: Receiver<TrackRemoteEvent>,
+    ) -> Self {
         Self {
             track,
+            receiver_id,
+            msg_tx,
             evt_rx: Mutex::new(evt_rx),
         }
     }
@@ -27,11 +40,10 @@ impl Track for TrackRemoteStaticRTP {
 
 #[async_trait::async_trait]
 impl TrackRemote for TrackRemoteStaticRTP {
-    async fn write_rtcp(
-        &self,
-        _packets: Vec<Box<dyn rtc::rtcp::Packet>>,
-    ) -> crate::error::Result<()> {
-        todo!()
+    async fn write_rtcp(&self, packets: Vec<Box<dyn rtc::rtcp::Packet>>) -> Result<()> {
+        self.msg_tx
+            .try_send(MessageInner::ReceiverRtcp(self.receiver_id, packets))
+            .map_err(|e| Error::Other(format!("{:?}", e)))
     }
 
     async fn poll(&self) -> Option<TrackRemoteEvent> {
diff --git a/src/peer_connection/driver.rs b/src/peer_connection/driver.rs
@@ -32,7 +32,7 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 /// Capacity of the internal driver message channel (WriteNotify, IceGathering, Close, …).
-pub(crate) const MESSAGE_INNER_CHANNEL_CAPACITY: usize = 64;
+pub(crate) const MESSAGE_INNER_CHANNEL_CAPACITY: usize = 256;
 
 /// Capacity of each data-channel event channel (OnOpen, OnMessage, OnClose, …).
 pub(crate) const DATA_CHANNEL_EVENT_CHANNEL_CAPACITY: usize = 256;
@@ -410,7 +410,12 @@ where
                     if let Some(track) = track {
                         let (evt_tx, evt_rx) = channel(TRACK_REMOTE_EVENT_CHANNEL_CAPACITY);
                         let track_remote: Arc<dyn TrackRemote> =
-                            Arc::new(TrackRemoteStaticRTP::new(track, evt_rx));
+                            Arc::new(TrackRemoteStaticRTP::new(
+                                track,
+                                init.receiver_id,
+                                self.inner.msg_tx.clone(),
+                                evt_rx,
+                            ));
 
                         {
                             let mut rtp_transceivers = self.inner.rtp_transceivers.lock().await;
@@ -512,12 +517,17 @@ where
 
     async fn handle_inner_message(&mut self, msg: MessageInner) -> bool {
         match msg {
-            /*MessageInner::SenderRtp(sender_id, packet) => {
+            MessageInner::SenderRtp(sender_id, packet) => {
                 let mut core = self.inner.core.lock().await;
                 if let Some(mut sender) = core.rtp_sender(sender_id) {
                     if let Err(err) = sender.write_rtp(packet) {
                         error!("Failed to send RTP: {}", err);
                     }
+                } else {
+                    error!(
+                        "Failed to send RTCP feedback due to unknown sender id {:?}",
+                        sender_id
+                    );
                 }
             }
             MessageInner::SenderRtcp(sender_id, rtcp_packets) => {
@@ -526,6 +536,11 @@ where
                     if let Err(err) = sender.write_rtcp(rtcp_packets) {
                         error!("Failed to send RTCP: {}", err);
                     }
+                } else {
+                    error!(
+                        "Failed to send RTCP feedback due to unknown sender id {:?}",
+                        sender_id
+                    );
                 }
             }
             MessageInner::ReceiverRtcp(receiver_id, rtcp_packets) => {
@@ -534,8 +549,13 @@ where
                     if let Err(err) = receiver.write_rtcp(rtcp_packets) {
                         error!("Failed to send RTCP feedback: {}", err);
                     }
+                } else {
+                    error!(
+                        "Failed to send RTCP feedback due to unknown receiver id {:?}",
+                        receiver_id
+                    );
                 }
-            }*/
+            }
             MessageInner::WriteNotify => {
                 //Do nothing, just want to wake up from futures::select! in order to poll_write
             }
diff --git a/src/peer_connection/mod.rs b/src/peer_connection/mod.rs
@@ -25,11 +25,14 @@ use rtc::data_channel::{RTCDataChannelId, RTCDataChannelInit};
 use rtc::peer_connection::RTCPeerConnectionBuilder;
 use rtc::peer_connection::configuration::{RTCAnswerOptions, RTCOfferOptions};
 use rtc::rtp_transceiver::rtp_sender::RtpCodecKind;
-use rtc::rtp_transceiver::{RTCRtpTransceiverId, RTCRtpTransceiverInit};
+use rtc::rtp_transceiver::{
+    RTCRtpReceiverId, RTCRtpSenderId, RTCRtpTransceiverId, RTCRtpTransceiverInit,
+};
 use rtc::sansio::Protocol;
 use rtc::shared::error::{Error, Result};
 use rtc::statistics::StatsSelector;
 use rtc::statistics::report::RTCStatsReport;
+use rtc::{rtcp, rtp};
 
 use crate::media_stream::track_local::static_rtp::TrackLocalStaticRTP;
 use crate::media_stream::track_remote::TrackRemoteEvent;
@@ -109,6 +112,9 @@ pub trait PeerConnectionEventHandler: Send + Sync + 'static {
 /// Unified inner message type for the peer connection driver
 #[derive(Debug)]
 pub(crate) enum MessageInner {
+    SenderRtp(RTCRtpSenderId, rtp::Packet),
+    SenderRtcp(RTCRtpSenderId, Vec<Box<dyn rtcp::Packet>>),
+    ReceiverRtcp(RTCRtpReceiverId, Vec<Box<dyn rtcp::Packet>>),
     WriteNotify,
     IceGathering,
     Close,
